/* Group Executable flow by Executors to reduce number of REST calls
194808
private Map<Optional<Executor>, List<ExecutableFlow>> getFlowToExecutorMap() {
    final HashMap<Optional<Executor>, List<ExecutableFlow>> exFlowMap =
        new HashMap<>();

    for (final Pair<ExecutionReference, ExecutableFlow> runningFlow : this.runningExecutions.get()
        .values()) {
      final ExecutionReference ref = runningFlow.getFirst();
      final ExecutableFlow flow = runningFlow.getSecond();
      final Optional<Executor> executor = ref.getExecutor();

      // We can set the next check time to prevent the checking of certain
      // flows.
      if (ref.getNextCheckTime() >= DateTime.now().getMillis()) {
        continue;
      }

      List<ExecutableFlow> flows = exFlowMap.get(executor);
      if (flows == null) {
        flows = new ArrayList<>();
        exFlowMap.put(executor, flows);
      }

      flows.add(flow);
    }

    return exFlowMap;
  }
590999
private boolean isFlowRunningHelper(final int projectId, final String flowId,
      final Collection<Pair<ExecutionReference, ExecutableFlow>> collection) {
    for (final Pair<ExecutionReference, ExecutableFlow> ref : collection) {
      if (ref.getSecond().getProjectId() == projectId
          && ref.getSecond().getFlowId().equals(flowId)) {
        return true;
      }
    }
    return false;
  }
553373
def ProcessHuntFlowDone(flow_obj, status_msg=None):
  """"""

  if not hunt.IsLegacyHunt(flow_obj.parent_hunt_id):
    hunt_obj = hunt.StopHuntIfCPUOrNetworkLimitsExceeded(
        flow_obj.parent_hunt_id)
    hunt.CompleteHuntIfExpirationTimeReached(hunt_obj)
    return

  hunt_urn = rdfvalue.RDFURN("hunts").Add(flow_obj.parent_hunt_id)
  client_urn = rdf_client.ClientURN(flow_obj.client_id)

  # Update the counter metrics separately from collections to minimize
  # contention.
  with aff4.FACTORY.Open(hunt_urn, mode="rw") as fd:
    # Legacy AFF4 code expects token to be set.
    fd.token = access_control.ACLToken(username=fd.creator)

    if flow_obj.num_replies_sent:
      fd.context.clients_with_results_count += 1

    fd.context.completed_clients_count += 1
    fd.context.results_count += flow_obj.num_replies_sent

    fd.GetRunner().SaveResourceUsage(flow_obj.client_id, status_msg)

  with aff4.FACTORY.Open(hunt_urn, mode="rw") as fd:
    # Legacy AFF4 code expects token to be set.
    fd.token = access_control.ACLToken(username=fd.creator)

    fd.RegisterCompletedClient(client_urn)
    if flow_obj.num_replies_sent:
      fd.RegisterClientWithResults(client_urn)

    fd.StopHuntIfAverageLimitsExceeded()
459587
private StringBuilder buildFlowURL(String cluster, String username,
      String batchDesc, String signature, int limit,
      List<String> flowResponseFilters, List<String> jobResponseFilters)
      throws IOException {
    StringBuilder urlStringBuilder = new StringBuilder();
    urlStringBuilder.append("http://");
    urlStringBuilder.append(apiHostname);
    urlStringBuilder.append(RestJSONResource.SLASH);
    urlStringBuilder.append(URL_PORTION_API_V1);
    urlStringBuilder.append(FLOW_API);
    urlStringBuilder.append(RestJSONResource.SLASH);
    urlStringBuilder.append(cluster);
    urlStringBuilder.append(RestJSONResource.SLASH);
    urlStringBuilder.append(username);
    urlStringBuilder.append(RestJSONResource.SLASH);
    urlStringBuilder.append(StringUtil.cleanseToken(batchDesc));
    if (StringUtils.isNotEmpty(signature)) {
      urlStringBuilder.append(RestJSONResource.SLASH);
      urlStringBuilder.append(signature);
    }
    urlStringBuilder.append(QUESTION_MARK);
    urlStringBuilder.append(LIMIT);
    urlStringBuilder.append(EQUAL_TO);
    urlStringBuilder.append(limit);
    if ((flowResponseFilters != null) && (flowResponseFilters.size() > 0)) {
      urlStringBuilder.append(AND);
      urlStringBuilder
          .append(StringUtil.buildParam("include", flowResponseFilters));
    }

    if ((jobResponseFilters != null) && (jobResponseFilters.size() > 0)) {
      urlStringBuilder.append(AND);
      urlStringBuilder
          .append(StringUtil.buildParam("includeJobField", jobResponseFilters));
    }

    return urlStringBuilder;
  }
693589
@SuppressWarnings("unchecked")
  public void updateExecutions() {
    this.updaterStage.set("Starting update all flows.");
    final Map<Optional<Executor>, List<ExecutableFlow>> exFlowMap = getFlowToExecutorMap();
    final ArrayList<ExecutableFlow> finalizeFlows =
        new ArrayList<>();

    for (final Map.Entry<Optional<Executor>, List<ExecutableFlow>> entry : exFlowMap
        .entrySet()) {

      final Optional<Executor> executorOption = entry.getKey();
      if (!executorOption.isPresent()) {
        for (final ExecutableFlow flow : entry.getValue()) {
          logger.warn("Finalizing execution " + flow.getExecutionId()
              + ". Executor id of this execution doesn't exist");
          finalizeFlows.add(flow);
        }
        continue;
      }
      final Executor executor = executorOption.get();

      this.updaterStage.set("Starting update flows on " + executor.getHost() + ":"
          + executor.getPort());

      Map<String, Object> results = null;
      try {
        results = this.apiGateway.updateExecutions(executor, entry.getValue());
      } catch (final ExecutorManagerException e) {
        handleException(entry, executor, e, finalizeFlows);
      }

      if (results != null) {
        final List<Map<String, Object>> executionUpdates =
            (List<Map<String, Object>>) results
                .get(ConnectorParams.RESPONSE_UPDATED_FLOWS);
        for (final Map<String, Object> updateMap : executionUpdates) {
          try {
            final ExecutableFlow flow = updateExecution(updateMap);

            this.updaterStage.set("Updated flow " + flow.getExecutionId());

            if (ExecutionControllerUtils.isFinished(flow)) {
              finalizeFlows.add(flow);
            }
          } catch (final ExecutorManagerException e) {
            final ExecutableFlow flow = e.getExecutableFlow();
            logger.error(e);

            if (flow != null) {
              logger.warn("Finalizing execution " + flow.getExecutionId());
              finalizeFlows.add(flow);
            }
          }
        }
      }
    }

    this.updaterStage.set("Finalizing " + finalizeFlows.size() + " error flows.");

    for (final ExecutableFlow flow : finalizeFlows) {
      this.executionFinalizer
          .finalizeFlow(flow, "Not running on the assigned executor (any more)", null);
    }

    this.updaterStage.set("Updated all active flows. Waiting for next round.");
  }
685477
function executeBatch(bulkOperation, batch, options, callback) {
  function resultHandler(err, result) {
    // Error is a driver related error not a bulk op error, terminate
    if (((err && err.driver) || (err && err.message)) && !(err instanceof MongoWriteConcernError)) {
      return handleCallback(callback, err);
    }

    // If we have and error
    if (err) err.ok = 0;
    if (err instanceof MongoWriteConcernError) {
      return handleMongoWriteConcernError(batch, bulkOperation.s.bulkResult, false, err, callback);
    }
    handleCallback(
      callback,
      null,
      mergeBatchResults(false, batch, bulkOperation.s.bulkResult, err, result)
    );
  }

  bulkOperation.finalOptionsHandler({ options, batch, resultHandler }, callback);
}
97910
function executePromiseGetters(getters) {
  var concurrent = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : 1;
  var stopped;
  var promise = new Promise(function (resolve, reject) {
    var r = [];
    var chunks = splitArray(getters, concurrent);
    var promise = Promise.resolve();
    chunks.forEach(function (chunk) {
      promise = promise.then(function (result) {
        if (result) {
          r.push.apply(r, _toConsumableArray(result));
        }

        if (stopped) {
          reject('stopped');
        } else {
          return Promise.all(chunk.map(function (v) {
            return v();
          }));
        }
      });
    });
    promise.then(function (result) {
      r.push.apply(r, _toConsumableArray(result));
      resolve(r);
    });
  });
  return {
    promise: promise,
    destroy: function destroy() {
      stopped = true;
    }
  };
}
7441
private void shutdownGracefully(final Iterator<EventExecutorGroup> iterator) {
        if (iterator.hasNext()) {
            EventExecutorGroup group = iterator.next();
            if (!group.isShuttingDown()) {
                group.shutdownGracefully().addListener(future -> {
                    if (!future.isSuccess()) {
                        //log.debug("shutdown of {} resulted in exception", group, future.cause());
                    }
                    shutdownGracefully(iterator);
                });
            }
        }
    }
1050545
protected static RedisFuture<Long> aggregateAsync(Map<?, ? extends CompletionStage<Long>> executions) {

        return new PipelinedRedisFuture<>(executions, objectPipelinedRedisFuture -> {
            AtomicLong result = new AtomicLong();
            for (CompletionStage<Long> future : executions.values()) {
                Long value = execute(() -> future.toCompletableFuture().get());
                if (value != null) {
                    result.getAndAdd(value);
                }
            }

            return result.get();
        });
    }
641815
private List<Integer> getRunningFlowsHelper(final int projectId, final String flowId,
      final Collection<Pair<ExecutionReference, ExecutableFlow>> collection) {
    final List<Integer> executionIds = new ArrayList<>();
    for (final Pair<ExecutionReference, ExecutableFlow> ref : collection) {
      if (ref.getSecond().getFlowId().equals(flowId)
          && ref.getSecond().getProjectId() == projectId) {
        executionIds.add(ref.getFirst().getExecId());
      }
    }
    return executionIds;
  }
691787
function prepareAndGetExecutor(target, out, operation) {
    target.privateIpAddress = target.privateIpAddress || target.ipAddress || target.ipaddress;
    var executor = platform.executor(config, target.privateIpAddress, os.platform(), logger);
    logger.info(operation);
    out.stdout(operation);
    return executor;
   }
510434
@Nullable
  private EventExecutorGroup createEventExecutorGroup(int size) {
    if (size <= 0) {
      return null;
    }

    ThreadFactory threadFactory = new ThreadFactory() {
      private final ThreadGroup threadGroup = new ThreadGroup(serviceName + "-executor-thread");
      private final AtomicLong count = new AtomicLong(0);

      @Override
      public Thread newThread(Runnable r) {
        Thread t = new Thread(threadGroup, r, String.format("%s-executor-%d", serviceName, count.getAndIncrement()));
        t.setDaemon(true);
        return t;
      }
    };

    UnorderedThreadPoolEventExecutor executor = new UnorderedThreadPoolEventExecutor(size, threadFactory,
                                                                                     rejectedExecutionHandler);
    if (execThreadKeepAliveSecs > 0) {
      executor.setKeepAliveTime(execThreadKeepAliveSecs, TimeUnit.SECONDS);
      executor.allowCoreThreadTimeOut(true);
    }
    return new NonStickyEventExecutorGroup(executor);
  }
150
@Override
  public List<ExecutableFlow> getRunningFlows() {
    final ArrayList<ExecutableFlow> flows = new ArrayList<>();
    try {
      getFlowsHelper(flows, this.executorLoader.fetchUnfinishedFlows().values());
    } catch (final ExecutorManagerException e) {
      logger.error("Failed to get running flows.", e);
    }
    return flows;
  }
717489
function processGroup(_records, _target, _params) {

    // extract targets
    var targets = [], record;
    for(var i = 0, l = _records.length; i < l; i++) {
      record = _records[i][_target];
      if(angular.isArray(record)) {
        targets.push.apply(targets, record);
      } else if(record) {
        targets.push(record);
      }
    }

    // populate targets
    if(targets.length > 0) {
      var promise = typeof targets[0].$type.$populate === 'function' ?
        targets[0].$type.$populate(targets, _params).$asPromise() :
        populate(targets, _params);

      if(promise) {
        return promise.then(function() {
          return targets;
        });
      }
    }

    return $q.when(targets);
  }
792166
def filter_last_write(client, groups, start):
    
    retry = get_retry(('ThrottlingException',))

    def process_group(group_set):
        matched = []
        for g in group_set:
            streams = retry(
                client.describe_log_streams,
                logGroupName=g['logGroupName'],
                orderBy='LastEventTime',
                limit=1, descending=True)
            if not streams.get('logStreams'):
                continue
            stream = streams['logStreams'][0]
            if stream['storedBytes'] == 0 and datetime.fromtimestamp(
                    stream['creationTime'] / 1000) > start:
                matched.append(g)
            elif 'lastIngestionTime' in stream and datetime.fromtimestamp(
                    stream['lastIngestionTime'] / 1000) > start:
                matched.append(g)
        return matched

    results = []

    with ThreadPoolExecutor(max_workers=3) as w:
        futures = {}
        for group_set in chunks(groups, 10):
            futures[w.submit(process_group, group_set)] = group_set

        for f in as_completed(futures):
            if f.exception():
                log.error(
                    "Error processing groupset:%s error:%s",
                    group_set,
                    f.exception())
            results.extend(f.result())

    return results
1306341
private void getFlowsHelper(final ArrayList<ExecutableFlow> flows,
      final Collection<Pair<ExecutionReference, ExecutableFlow>> collection) {
    collection.stream().forEach(ref -> flows.add(ref.getSecond()));
  }
1362900
private void optimize(long fileLength) {

		final int availableProcessors = Runtime.getRuntime().availableProcessors();

		final long free = Runtime.getRuntime().freeMemory() / 2;

		int workerSize = availableProcessors / 2;
		if (workerSize < 2) {
			workerSize = 2;
		}

		long bufferSize = free / workerSize;
		if (bufferSize > 1 * 1024 * 1024) {
			bufferSize = 1 * 1024 * 1024;
		}

		long blockSize = fileLength / workerSize;
		if (blockSize > 1 * 1024 * 1024) {
			blockSize = 1 * 1024 * 1024;
		}
		int iBlockSize = (int) blockSize;

		if (bufferSize > blockSize) {
			bufferSize = blockSize;
		}
		int iBufferSize = (int) bufferSize;

		this.setBlockSize(iBlockSize);
		this.setMaxNumOfThreads(workerSize);
		this.setBufferSizePerWorker(iBufferSize);
		this.setSubBufferSize(256);

	}