Sends an {@link VortexAggregateFunction} and its {@link VortexFunction} to a
{@link org.apache.reef.vortex.evaluator.VortexWorker}.
496172
<TInput, TOutput> void sendAggregateFunction(final int aggregateFunctionId,
                                               final VortexAggregateFunction<TOutput> aggregateFunction,
                                               final VortexFunction<TInput, TOutput> function,
                                               final VortexAggregatePolicy policy) {
    final TaskletAggregationRequest<TInput, TOutput> taskletAggregationRequest =
        new TaskletAggregationRequest<>(aggregateFunctionId, aggregateFunction, function, policy);

    // The send is synchronous such that we make sure that the aggregate function is sent to the
    // target worker before attempting to launch an aggregateable tasklet on it.
    vortexRequestor.send(reefTask, taskletAggregationRequest);
  }
509758
@Beta
    public static <T, U, R> BiFunction<T, U, R> sf(final Object mutex, final BiFunction<T, U, R> biFunction) {
        N.checkArgNotNull(mutex, "mutex");
        N.checkArgNotNull(biFunction, "biFunction");

        return new BiFunction<T, U, R>() {
            @Override
            public R apply(T t, U u) {
                synchronized (mutex) {
                    return biFunction.apply(t, u);
                }
            }
        };
    }
429555
private void aggregateTasklets(final AggregateTriggerType type) {
    final List<WorkerToMasterReport> workerToMasterReports = new ArrayList<>();
    final List<Object> results = new ArrayList<>();
    final List<Integer> aggregatedTasklets = new ArrayList<>();

    // Synchronization to prevent duplication of work on the same aggregation function on the same worker.
    synchronized (stateLock) {
      switch(type) {
      case ALARM:
        aggregateTasklets(workerToMasterReports, results, aggregatedTasklets);
        break;
      case COUNT:
        if (!aggregateOnCount()) {
          return;
        }

        aggregateTasklets(workerToMasterReports, results, aggregatedTasklets);
        break;
      default:
        throw new RuntimeException("Unexpected aggregate type.");
      }
    }

    if (!results.isEmpty()) {
      // Run the aggregation function.
      try {
        final Object aggregationResult = taskletAggregationRequest.executeAggregation(results);
        workerToMasterReports.add(new TaskletAggregationResultReport(aggregatedTasklets, aggregationResult));
      } catch (final Exception e) {
        workerToMasterReports.add(new TaskletAggregationFailureReport(aggregatedTasklets, e));
      }
    }

    // Add to worker report only if there is something to report back.
    if (!workerToMasterReports.isEmpty()) {
      workerReportsQueue.addLast(kryoUtils.serialize(new WorkerToMasterReports(workerToMasterReports)));
      heartBeatTriggerManager.triggerHeartBeat();
    }
  }
642272
protected final void sendObjectToSocket(final Object objectToSend) {
        this.sendObjectToSocket(objectToSend, new WriteCallback() {

            @Override
            public void writeSuccess() {
                ServerJSONWebSocketAdapter.LOGGER.debug("Send data to socket: {}", objectToSend);
            }

            @Override
            public void writeFailed(Throwable x) {
                ServerJSONWebSocketAdapter.LOGGER.error("Error sending message to socket", x);
            }
        });
    }
1049371
private void sendQueuedInvocations() {
        while (true) {
            MethodInvocation<?> currentRPC = queuedRpcList.poll();
            if (currentRPC == null) {
                return;
            }

            try {
                connector.executeAsyncMethod(currentRPC.getProxy(),
                                             currentRPC.getMethod(),
                                             currentRPC.getArgs(),
                                             currentRPC.getFuture());
            } catch (JoynrRuntimeException e) {
                currentRPC.getFuture().onFailure(e);
            } catch (Exception e) {
                currentRPC.getFuture().onFailure(new JoynrRuntimeException(e));
            }

        }

    }
596903
private static Runnable rejectionPropagatingRunnable(
      final AbstractFuture<?> outputFuture,
      final Runnable delegateTask,
      final Executor delegateExecutor) {
    return new Runnable() {
      @Override public void run() {
        final AtomicBoolean thrownFromDelegate = new AtomicBoolean(true);
        try {
          delegateExecutor.execute(new Runnable() {
            @Override public void run() {
              thrownFromDelegate.set(false);
              delegateTask.run();
            }
          });
        } catch (RejectedExecutionException e) {
          if (thrownFromDelegate.get()) {
            // wrap exception?
            outputFuture.setException(e);
          }
          // otherwise it must have been thrown from a transitive call and the delegate runnable
          // should have handled it.
        }
      }
    };
  }
44655
void addWorker(final VortexWorkerManager vortexWorkerManager) {
    lock.lock();
    try {
      if (!terminated) {
        if (!removedBeforeAddedWorkers.contains(vortexWorkerManager.getId())) {
          this.runningWorkers.put(vortexWorkerManager.getId(), vortexWorkerManager);
          this.schedulingPolicy.workerAdded(vortexWorkerManager);
          this.workerAggregateFunctionMap.put(vortexWorkerManager.getId(), new HashSet<Integer>());

          // Notify (possibly) waiting scheduler
          noWorkerOrResource.signal();
        }
      } else {
        // Terminate the worker
        vortexWorkerManager.terminate();
      }
    } finally {
      lock.unlock();
    }
  }
415597
public static void useNioTransport(NettyServerBuilder builder) {
    builder.channelType(NioServerSocketChannel.class);
    builder
        .bossEventLoopGroupPool(SharedResourcePool.forResource(Utils.NIO_BOSS_EVENT_LOOP_GROUP));
    builder
        .workerEventLoopGroupPool(
            SharedResourcePool.forResource(Utils.NIO_WORKER_EVENT_LOOP_GROUP));
  }
491147
@Override
  public <TInput, TOutput> VortexFuture<TOutput>
      enqueueTasklet(final VortexFunction<TInput, TOutput> function, final TInput input,
                     final Optional<FutureCallback<TOutput>> callback) {
    // TODO[REEF-500]: Simple duplicate Vortex Tasklet launch.
    final VortexFuture<TOutput> vortexFuture;
    final int id = taskletIdCounter.getAndIncrement();
    if (callback.isPresent()) {
      vortexFuture = new VortexFuture<>(executor, this, id, callback.get());
    } else {
      vortexFuture = new VortexFuture<>(executor, this, id);
    }

    final Tasklet tasklet = new Tasklet<>(id, Optional.<Integer>empty(), function, input, vortexFuture);
    putDelegate(Collections.singletonList(tasklet), vortexFuture);
    this.pendingTasklets.addLast(tasklet);

    return vortexFuture;
  }
492962
@Override
    protected void sendMessage(@NonNull INDArray message, int iterationNumber, int epochNumber) {
        // here we'll send our stuff to other executores over the wire
        // and let's pray for udp broadcast availability

        // Send this message away
        // FIXME: do something with unsafe duplication, which is bad and used ONLY for local spark
        try (MemoryWorkspace wsO = Nd4j.getMemoryManager().scopeOutOfWorkspaces()) {
            long updateId = updatesCounter.getAndIncrement();

            val m = message.unsafeDuplication();
            ModelParameterServer.getInstance().sendUpdate(m, iterationNumber, epochNumber);
        }


        // heere we update local queue
        super.sendMessage(message, iterationNumber, epochNumber);
    }
1155698
@Override
    public void sendFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendFlushedMessage", new Object[] { streamID });

        // This flush should be broadcast to all downstream neighbors 
        // for this cell as it is the result of a startFlush()

        // This is a bit of a kludge since we may be sending flushes to cells
        // which have no information about a stream.
        HashMap allPubSubOutputHandlers = _destination.getAllPubSubOutputHandlers();
        try
        {
            Iterator itr = allPubSubOutputHandlers.values().iterator();

            while (itr.hasNext())
            {
                // Get the appropriate target cellule and forward the message
                PubSubOutputHandler handler = (PubSubOutputHandler) itr.next();

                // Note that the null Cellule we pass in here is ignored
                // as each OutputHandler knows its targetCellule 
                handler.sendFlushedMessage(null, streamID);

                // Also, tell the handler to remove any information for this
                // stream since it's just been flushed.
                handler.removeStream(streamID);
            }
        } finally
        {
            // By calling the getAllPubSubOutputHandlers it will lock the 
            // handlers
            _destination.unlockPubsubOutputHandlers();
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendFlushedMessage");
    }
715359
private void executeAggregateTasklet(final ExecutorService commandExecutor,
                                       final MasterToWorkerRequest masterToWorkerRequest) {
    final TaskletAggregateExecutionRequest taskletAggregateExecutionRequest =
        (TaskletAggregateExecutionRequest) masterToWorkerRequest;

    assert aggregates.containsKey(taskletAggregateExecutionRequest.getAggregateFunctionId());

    final AggregateContainer aggregateContainer =
        aggregates.get(taskletAggregateExecutionRequest.getAggregateFunctionId());
    final TaskletAggregationRequest aggregationRequest = aggregateContainer.getTaskletAggregationRequest();

    commandExecutor.submit(new Runnable() {
      @Override
      public void run() {
        try {
          aggregateContainer.scheduleTasklet(taskletAggregateExecutionRequest.getTaskletId());
          final Object result = aggregationRequest.executeFunction(taskletAggregateExecutionRequest.getInput());
          aggregateContainer.taskletComplete(taskletAggregateExecutionRequest.getTaskletId(), result);
        } catch (final Exception e) {
          aggregateContainer.taskletFailed(taskletAggregateExecutionRequest.getTaskletId(), e);
        }
      }
    });
  }
72808
private synchronized VortexFutureDelegate fetchDelegate(final List<Integer> taskletIds) {
    VortexFutureDelegate delegate = null;
    for (final int taskletId : taskletIds) {
      final VortexFutureDelegate currDelegate = taskletFutureMap.remove(taskletId);
      if (currDelegate == null) {
        // TODO[JIRA REEF-500]: Consider duplicate tasklets.
        throw new RuntimeException("Tasklet should only be removed once.");
      }

      if (delegate == null) {
        delegate = currDelegate;
      } else {
        assert delegate == currDelegate;
      }
    }

    return delegate;
  }
952452
@Override
  public void sendRegistrationTerminationAnswer(JRegistrationTerminationAnswer answer)
      throws InternalException, IllegalDiameterStateException, RouteException, OverloadException {
    send(Event.Type.SEND_MESSAGE, null, answer);
  }
1133223
@Override
  public void start(final VortexThreadPool vortexThreadPool) {
    final Vector<Integer> inputVector = new Vector<>();
    for (int i = 0; i < dimension; i++) {
      inputVector.add(i);
    }

    final List<VortexFuture<Integer>> futures = new ArrayList<>();
    final AddOneFunction addOneFunction = new AddOneFunction();
    for (final int i : inputVector) {
      futures.add(vortexThreadPool.submit(addOneFunction, i));
    }

    final Vector<Integer> outputVector = new Vector<>();
    for (final VortexFuture<Integer> future : futures) {
      try {
        outputVector.add(future.get());
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }

    System.out.println("RESULT:");
    System.out.println(outputVector);
  }
835002
private Object invokeImpl(Object thiz, String name, Object... args)
            throws ScriptException, NoSuchMethodException {
        if (name == null) {
            throw new NullPointerException("method name is null");
        }

        try {
            if (thiz != null) {
                return InvokerHelper.invokeMethod(thiz, name, args);
            } else {
                return callGlobal(name, args);
            }
        } catch (MissingMethodException mme) {
            throw new NoSuchMethodException(mme.getMessage());
        } catch (Exception e) {
            throw new ScriptException(e);
        }
    }
493669
protected void sendCollectorDataToAnalyzerIfPresent(Serializable data) {
		if (analyzerStatusQueue != null) {
			logger.finer("Sending collector partition data: " + data + " to analyzer queue: " + analyzerStatusQueue);
			PartitionDataWrapper dataWrapper = new PartitionDataWrapper();
			dataWrapper.setCollectorData(data);
			dataWrapper.setEventType(PartitionEventType.ANALYZE_COLLECTOR_DATA);
			analyzerStatusQueue.add(dataWrapper);
		} else {
			logger.fine("Analyzer not configured.");
		}
	}