jgapclientgp.java

来自「jGAp 遗传算法 提不错的一款软件 最新的更新」· Java 代码 · 共 1,135 行 · 第 1/3 页

                m_gcmed.removeMessage(key);
                m_objects.getResults().remove(key);
              }
            }
            try {
              try {
                if (m_list) {
                  // List existing requests and results with extended information.
                  // -------------------------------------------------------------
                  listRequests();
                  listResults();
                }
                if (!m_receiveOnly && !m_no_evolution) {
                  // Initialize evolution.
                  // ---------------------
                  IClientEvolveStrategyGP clientEvolver = m_gridConfig.
                      getClientEvolveStrategy();
                  if (clientEvolver != null) {
                    clientEvolver.initialize(m_gcmed, getConfiguration(),
                        m_gridConfig.getClientFeedback());
                  }
                }
                if (!m_no_evolution) {
                  // Do the evolution.
                  // -----------------
                  beforeEvolve(m_gcmed);
                  evolve(m_gcmed, m_receiveOnly);
                  afterEvolve(m_gcmed);
                }
              } catch (Exception ex) {
                log.error("Error", ex);
                throw ex;
              }
            } finally {
              Throwable t = null;
              try {
                try {
                  m_gcmed.stop();
                } catch (Throwable t1) {
                  t = t1;
                }
              } finally {
                afterStopped(t);
                break;
              }
            }
          } catch (Exception ex) {
            log.fatal("Unpredicted error", ex);
            m_gridConfig.getClientFeedback().error("Error while doing the work",
                ex);
            onError(ex);
            try {
//              m_gcmed.disconnect();
            } catch (Exception ex1) {
              log.warn("Precautios disconnect failed.", ex1);
            }
            sleep(10000);
          }
        } while (true);
        if (!m_endless) {
          break;
        }
        else {
          log.info("Starting again after a short break...");
          sleep(15000);
        }
      } while (true);
    } catch (InterruptedException iex) {
      // Thread interrupted.
      // -------------------
      log.fatal("Thread was interrupted", iex);
      try {
        m_gcmed.disconnect();
      } catch (Exception ex) {
        log.warn("Disconnect after interruption failed", ex);
      }
    } catch (Throwable t) {
      t.printStackTrace();
    }
    log.info("Stopping client");
  }

  protected JGAPRequestGP[] sendWorkRequests(int a_evolutionIndex,
      IClientEvolveStrategyGP evolver, IRequestSplitStrategyGP splitter,
      IClientFeedbackGP feedback)
      throws Exception {
    JGAPRequestGP[] workRequests = null;
    if (beforeGenerateWorkRequests()) {
      log.info("Beginning evolution cycle " + a_evolutionIndex);
      try {
//      m_clientEvolveStrategy.beforeGenerateWorkResults();
        workRequests = evolver.generateWorkRequests(m_workReq, splitter, null);
        feedback.setProgressMaximum(0);
        feedback.setProgressMaximum(workRequests.length - 1);
        for (int i = 0; i < workRequests.length; i++) {
          presetPopulation(workRequests[i]);
        }
        if (beforeSendWorkRequests(workRequests)) {
          /**@todo merge previous results in req.getPopulation()*/
          if (!m_no_comm) {
            try {
              sendWorkRequests(workRequests);
              return workRequests;
            } catch (Exception ex) {
              throw new WorkRequestsSendException(ex, workRequests);
            }
          }
          else {
            return workRequests;
          }
        }
        else {
          return null;
        }
      } catch (Exception ex) {
        ex.printStackTrace();
        throw ex;
      }
    }
    else {
      return null;
    }
  }

  protected void sendWorkRequests(JGAPRequestGP[] a_workList)
      throws Exception {
    // Send work requests.
    // -------------------
    for (int i = 0; i < a_workList.length; i++) {
      JGAPRequestGP req = a_workList[i];
      GPPopulation pop = req.getPopulation();
      if (pop == null || pop.isFirstEmpty()) {
        log.debug("Population to send to worker is empty!");
      }
      m_gridConfig.getClientFeedback().sendingFragmentRequest(req);
      MessageContext context = new MessageContext(MODULE_CS,
          CONTEXT_WORK_REQUEST, CONTEXT_ID_EMPTY);
      m_gcmed.send(new GridMessageWorkRequest(req), context, null);
      if (isInterrupted()) {
        break;
      }
    }
  }

  protected void receiveWorkResults(JGAPRequestGP[] workList)
      throws Exception {
    log.info("Receiving work results...");
    IClientFeedbackGP feedback = m_gridConfig.getClientFeedback();
    // Receive work results.
    // ---------------------
    int size;
    if (workList == null) {
      size = -1;
    }
    else {
      size = workList.length;
    }
    if (m_WANMode && size < 1) {
      // First, get a list of all work results.
      // --------------------------------------
      MessageContext context = new MessageContext(MODULE_WS,
          /**@todo later: SC*/
          CONTEXT_WORK_RESULT, CONTEXT_ID_EMPTY);
      List requests = m_gcmed.listResults(context, null, null);
      // Then, iterate over them and receive one after another.
      // ------------------------------------------------------
      int i = 0;
      for (Object request : requests) {
        feedback.setProgressValue(i);
        i++;
        JGAPResultGP result = receiveWorkResult(request, feedback, true);
        if (result != null) {
          IGPProgram best = result.getPopulation().determineFittestProgram();
          log.info("Result received: " +
                   best.getFitnessValue());
          resultReceived(best);
          MasterInfo worker = result.getWorkerInfo();
          if (worker != null) {
            log.info(" Worker IP " + worker.m_IPAddress + ", host " +
                     worker.m_name);
          }
          // Store result to disk.
          // ---------------------
          String filename = "result_" + getRunID() + "_" + result.getID() +
              "_" +
              result.getSessionName() + "_" + result.getChunk();
          writeToFile(best, m_workDir, filename);
          // Now remove the result from the online store.
          // --------------------------------------------
          /**@todo do this here explicitely and not in receiveWorkResult*/
        }
      }
    }
    else {
      for (int i = 0; i < size; i++) {
        feedback.setProgressValue(i + workList.length);
        receiveWorkResult(workList, feedback);
        if (this.isInterrupted()) {
          break;
        }
      }
    }
  }

  private JGAPResultGP receiveWorkResult(Object a_result,
      IClientFeedbackGP feedback, boolean a_remove)
      throws Exception {
    MessageContext context = new MessageContext(MODULE_WS /**@todo later: SC*/,
        CONTEXT_WORK_RESULT, a_result);
    GridMessageWorkResult gmwr = (GridMessageWorkResult) m_gcmed.
        getGridMessage(context, null, TIMEOUT_SECONDS, WAITTIME_SECONDS,
                       a_remove);
    if (gmwr == null) {
      throw new WorkResultNotFoundException();
    }
    else {
      String s = " ";
      if (a_remove) {
        s += "and removed from WAN";
      }
      log.info("Work result received" + s);
    }
    JGAPResultGP workResult = (JGAPResultGP) gmwr.getWorkResult();
    m_gridConfig.getClientEvolveStrategy().resultReceived(workResult);
    int idx = workResult.getChunk();
    // Fire listener.
    // --------------
    feedback.receivedFragmentResult(null, workResult, idx);
    return workResult;
  }

  private JGAPResultGP receiveWorkResult(JGAPRequestGP[] workList,
      IClientFeedbackGP feedback)
      throws Exception {
    /**@todo make this asynchronous with fall-back and reconnect!*/
    MessageContext context = new MessageContext(MODULE_WS /**@todo later: SC*/,
        CONTEXT_WORK_RESULT, CONTEXT_ID_EMPTY);
    GridMessageWorkResult gmwr = (GridMessageWorkResult) m_gcmed.
        getGridMessage(context, null, TIMEOUT_SECONDS, WAITTIME_SECONDS, true);
    if (gmwr == null) {
      throw new NoWorkResultsFoundException();
    }
    else {
      log.info("Work result received!");
    }
    JGAPResultGP workResult = (JGAPResultGP) gmwr.getWorkResult();
    m_gridConfig.getClientEvolveStrategy().resultReceived(workResult);
    int idx = workResult.getChunk();
    // Fire listener.
    // --------------
    JGAPRequestGP req;
    if (workList == null || workList.length < 1) {
      req = null;
    }
    else {
      req = workList[idx];
    }
    feedback.receivedFragmentResult(req, workResult, idx);
    return workResult;
  }

  /**
   * If necessary: override to implement your evolution cycle individually.
   *
   * @param a_gcmed the GridClient mediator
   * @param a_receiveOnly false: Don't send any work requests, just receive
   * results from former evolutions
   *
   * @throws Exception
   */
  protected void evolve(IGridClientMediator a_gcmed, boolean a_receiveOnly)
      throws Exception {
    // Do the complete evolution cycle until end.
    // ------------------------------------------
    IClientFeedbackGP feedback = m_gridConfig.getClientFeedback();
    feedback.beginWork();
    IClientEvolveStrategyGP evolver = m_gridConfig.getClientEvolveStrategy();
    IRequestSplitStrategyGP splitter = m_gridConfig.getRequestSplitStrategy();
    int evolutionIndex = 0;
    do {
      JGAPRequestGP[] workRequests = null;
      boolean deferRequests = false;
      if (!a_receiveOnly) {
        try {
          // Care that not too much work requests are online, do a listing
          // from time to time. If enough requests already there, don't create
          // them any more.
          // -----------------------------------------------------------------
          long lastListing = m_objects.getLastListingRequestsMillis();
          long current = System.currentTimeMillis();
          if (current - lastListing > 60 * 60 * 1) { //60 Seconds * 60 Minutes * 1 Hour
            // Do a listing again after 60 minutes or more.
            // --------------------------------------------
            MessageContext context = new MessageContext(MODULE_CS,
                CONTEXT_WORK_REQUEST, CONTEXT_ID_EMPTY);
            List requests = a_gcmed.listRequests(context, null, null);
            m_objects.setLastListingRequestsMillis(current);
            m_persister.save();
            if (requests != null && requests.size() > 100) {
              deferRequests = true;
              log.info("Deferring creating and sending further requests"
                       + ", maximum reached ("
                       + requests.size() + " found).");
            }
            if (requests != null && requests.size() > 0) {
              // Remove requests from database that are not in list any more.
              // ------------------------------------------------------------
              Map foundKeys = new HashMap();
              Object first = requests.get(0);
              if (String.class.isAssignableFrom(first.getClass())) {
                // Requests of type String can be handled directly.
                // ------------------------------------------------
                for (Object key : requests) {
                  foundKeys.put(key, "");
                }
              }
              else {
                // Requests of type that sub classes have to handle.
                // -------------------------------------------------
                for (Object obj : requests) {
                  String key = getKeyFromObject(obj);
                  if (key != null) {
                    foundKeys.put(key, "");
                  }
                }
              }
              removeEntries(foundKeys, m_objects.getRequests());
            }
          }
          /**@todo do the same for results*/
          if (!deferRequests) {
            workRequests = sendWorkRequests(evolutionIndex, evolver, splitter,
                feedback);
          }
          else {
            // Defer creating and sending additional requests.
            // -----------------------------------------------
          }
        } catch (WorkRequestsSendException wex) {
          errorOnSendWorkRequests(wex.getCause(), wex.getWorkRequests());
//        } catch (UploadFailedException uex) {
//          errorOnSendWorkRequests(uex, null);
//          throw uex;
//        } catch (ListingFailedException lex) {
//          errorOnSendWorkRequests(lex, null);
//          throw lex;
        }
        if (!deferRequests && !afterSendWorkRequests(workRequests)) {
          break;
        }
      }
      if (this.isInterrupted()) {
        break;
      }
      if (!deferRequests && !a_receiveOnly) {
        evolver.afterWorkRequestsSent();
      }
      if (!m_no_comm) {
        try {
          receiveWorkResults(workRequests);
        } catch (NoWorkResultsFoundException nwr) {
          log.info("No work results found.");
        }
      }
      if (!a_receiveOnly && !m_no_evolution) {
        evolver.evolve();
        // Fire listener that one evolution cycle is complete.
        // ---------------------------------------------------
        feedback.completeFrame(evolutionIndex);
        evolutionIndex++;
        // Check if evolution is finished.
        // -------------------------------
        if (evolver.isEvolutionFinished(evolutionIndex)) {
          evolver.onFinished();
          break;
        }
      }
    } while (true);
    try {