Send an invocation directly to each host's CI mailbox.
This ONLY works for NT procedures.
Track responses and complete the returned future when they're all accounted for.
345499
protected CompletableFuture<Map<Integer,ClientResponse>> callAllNodeNTProcedure(String procName, Object... params) {
        // only one of these at a time
        if (!m_outstandingAllHostProc.compareAndSet(false, true)) {
            throw new VoltAbortException(new IllegalStateException("Only one AllNodeNTProcedure operation can be running at a time."));
        }

        StoredProcedureInvocation invocation = new StoredProcedureInvocation();
        invocation.setProcName(procName);
        invocation.setParams(params);
        invocation.setClientHandle(m_id);

        final Iv2InitiateTaskMessage workRequest =
                new Iv2InitiateTaskMessage(m_mailbox.getHSId(),
                                           m_mailbox.getHSId(),
                                           TransactionInfoBaseMessage.UNUSED_TRUNC_HANDLE,
                                           m_id,
                                           m_id,
                                           true,
                                           false,
                                           invocation,
                                           m_id,
                                           ClientInterface.NT_REMOTE_PROC_CID,
                                           false);

        m_allHostFut = new CompletableFuture<>();
        m_allHostResponses = new HashMap<>();

        // hold this lock while getting the count of live nodes
        // also held when
        long[] hsids;
        synchronized(m_allHostCallbackLock) {
            // collect the set of live client interface mailbox ids
            m_outstandingAllHostProcedureHostIds = VoltDB.instance().getHostMessenger().getLiveHostIds();
            // convert host ids to hsids
            hsids = m_outstandingAllHostProcedureHostIds.stream()
                    .mapToLong(hostId -> CoreUtils.getHSIdFromHostAndSite(hostId, HostMessenger.CLIENT_INTERFACE_SITE_ID))
                    .toArray();
        }

        // send the invocation to all live nodes
        // n.b. can't combine this step with above because sometimes the callbacks comeback so fast
        //  you get a concurrent modification exception
        for (long hsid : hsids) {
            m_mailbox.send(hsid, workRequest);
        }

        return m_allHostFut;
    }
506014
def send_response(self, transaction):
        
        host, port = transaction.request.source
        key_token = hash(str(host) + str(port) + str(transaction.request.token))
        if key_token in self._relations:
            if transaction.response.code == defines.Codes.CONTENT.number:
                if transaction.resource is not None and transaction.resource.observable:

                    transaction.response.observe = transaction.resource.observe_count
                    self._relations[key_token].allowed = True
                    self._relations[key_token].transaction = transaction
                    self._relations[key_token].timestamp = time.time()
                else:
                    del self._relations[key_token]
            elif transaction.response.code >= defines.Codes.ERROR_LOWER_BOUND:
                del self._relations[key_token]
        return transaction
211827
def _inbound(self, From, to, protocol, udp_source=None):
        
        # Verify stuff!

        self.verifyCertificateAllowed(to, From)
        return self.service.verifyHook(From, to, protocol
                                       ).addCallback(self._inboundimpl,
                                                     From,
                                                     to,
                                                     protocol,
                                                     udp_source).addErrback(
            lambda f: f.trap(KeyError) and dict(listeners=[]))
56749
def sendintr(self):
        ''' '''

        n, byte = self.ptyproc.sendintr()
        self._log_control(byte)
865301
def send(x, inter=0, loop=0, count=None, verbose=None, realtime=None, *args, **kargs):
    """"""
    __gen_send(conf.L3socket(*args, **kargs), x, inter=inter, loop=loop, count=count,verbose=verbose, realtime=realtime)
868812
def extern_generator_send(self, context_handle, func, arg):
    """"""
    c = self._ffi.from_handle(context_handle)
    response = self._ffi.new('PyGeneratorResponse*')
    try:
      res = c.from_value(func[0]).send(c.from_value(arg[0]))
      if isinstance(res, Get):
        # Get.
        response.tag = self._lib.Get
        response.get = (
            TypeId(c.to_id(res.product)),
            c.to_value(res.subject),
            c.identify(res.subject),
          )
      elif type(res) in (tuple, list):
        # GetMulti.
        response.tag = self._lib.GetMulti
        response.get_multi = (
            c.type_ids_buf([TypeId(c.to_id(g.product)) for g in res]),
            c.vals_buf([c.to_value(g.subject) for g in res]),
            c.identities_buf([c.identify(g.subject) for g in res]),
          )
      else:
        # Break.
        response.tag = self._lib.Broke
        response.broke = (c.to_value(res),)
    except Exception as e:
      # Throw.
      response.tag = self._lib.Throw
      val = e
      val._formatted_exc = traceback.format_exc()
      response.throw = (c.to_value(val),)

    return response[0]
740942
def emit(self, event, data=None, room=None, skip_sid=None, namespace=None,
             callback=None):
        
        return self.server.emit(event, data=data, room=room, skip_sid=skip_sid,
                                namespace=namespace or self.namespace,
                                callback=callback)
69666
def subscribe_inbox
      # double-check locking to reduce synchronization
      if @subject_id.nil?
        # nats lazy-load needs to be outside the synchronized block
        client = nats
        @lock.synchronize do
          if @subject_id.nil?
            @subject_id = client.subscribe("#{@inbox_name}.>") do |message, _, subject|
              @handled_response = true
              handle_response(message, subject)
            end
          end
        end
      end
    end
287732
def invoke(invocation):
    
    (invocation.module_path,
     invocation.module_source) = get_module_data(invocation.module_name)
    planner = _get_planner(invocation)

    if invocation.wrap_async:
        response = _invoke_async_task(invocation, planner)
    elif planner.should_fork():
        response = _invoke_isolated_task(invocation, planner)
    else:
        _propagate_deps(invocation, planner, invocation.connection.context)
        response = invocation.connection.get_chain().call(
            ansible_mitogen.target.run_module,
            kwargs=planner.get_kwargs(),
        )

    return invocation.action._postprocess_response(response)
1030156
@Override
  public void invoke(InboxAmp mailbox, StubAmp actorDeliver)
  {
    // actorDeliver.queryError(_headers, _from, _qid, _exn);
    
    StubAmp actorMessage = _serviceRef.stub();
    
    // kraken/210a vs baratine/2245
    StubAmp actorInvoke = actorDeliver.worker(actorMessage);
    
    // actorInvoke.queryError(_headers, actorMessage, _qid, _exn);
    actorInvoke.queryError(_headers, _from, _qid, _exn);
  }
924048
def invoke(self):
        
        args = self._get_args()
        self._action(*args)
        for callback in self._post_callbacks:
            callback()
959352
def sendmail_sync(
        self,
        sender: str,
        recipients: RecipientsType,
        message: Union[str, bytes],
        mail_options: Iterable[str] = None,
        rcpt_options: Iterable[str] = None,
        timeout: DefaultNumType = _default,
    ) -> SendmailResponseType:
        
        return self._run_sync(
            self.sendmail,
            sender,
            recipients,
            message,
            mail_options=mail_options,
            rcpt_options=rcpt_options,
            timeout=timeout,
        )
992205
def send(r, pools=None):
    """"""

    if pools:
        r._pools = pools

    r.send()

    return r.response
320877
private void sendCommand(Collection<Address> targets, ReplicableCommand command, long requestId,
                            DeliverOrder deliverOrder, boolean rsvp, boolean checkView) {
      Objects.requireNonNull(targets);
      Message message = new Message();
      marshallRequest(message, command, requestId);
      setMessageFlags(message, deliverOrder, rsvp, true);

      if (deliverOrder == DeliverOrder.TOTAL) {
         message.dest(new AnycastAddress(toJGroupsAddressList(targets)));
         send(message);
      } else {
         Message copy = message;
         for (Iterator<Address> it = targets.iterator(); it.hasNext(); ) {
            Address address = it.next();

            if (checkView && !clusterView.contains(address))
               continue;

            if (address.equals(getAddress()))
               continue;

            copy.dest(toJGroupsAddress(address));
            send(copy);

            // Send a different Message instance to each target
            if (it.hasNext()) {
               copy = copy.copy(true);
            }
         }
      }
   }
6442
protected boolean call(Object... paramListIn) {
        m_perCallStats = m_statsCollector.beginProcedure();

        // if we're keeping track, calculate parameter size
        if (m_perCallStats.samplingProcedure()) {
            ParameterSet params = ParameterSet.fromArrayNoCopy(paramListIn);
            m_perCallStats.setParameterSize(params.getSerializedSize());
        }

        ClientResponseImpl response = coreCall(paramListIn);

        // null response means this procedure isn't over and has some async component
        if (response == null) {
            return false;
        }

        // if the whole call is done (no async bits)

        // if we're keeping track, calculate result size
        if (m_perCallStats.samplingProcedure()) {
            m_perCallStats.setResultSize(response.getResults());
        }
        m_statsCollector.endProcedure(response.getStatus() == ClientResponse.USER_ABORT,
                                      (response.getStatus() != ClientResponse.USER_ABORT) &&
                                      (response.getStatus() != ClientResponse.SUCCESS),
                                      m_perCallStats);
        // allow the GC to collect per-call stats if this proc isn't called for a while
        m_perCallStats = null;

        // send the response to caller
        // must be done as IRM to CI mailbox for backpressure accounting
        response.setClientHandle(m_clientHandle);
        InitiateResponseMessage irm = InitiateResponseMessage.messageForNTProcResponse(m_ciHandle,
                                                                                       m_ccxn.connectionId(),
                                                                                       response);
        m_mailbox.deliver(irm);

        // remove record of this procedure in NTPS
        // only done if procedure is really done
        m_ntProcService.handleNTProcEnd(this);

        return true;
    }
962335
public void sendRequest (
        int invOid, int invCode, int methodId, Object[] args, Transport transport)
    {
        if (_clobj == null) {
            log.warning("Dropping invocation request on shutdown director", "code", invCode,
                        "methodId", methodId);
            return;
        }

        // configure any invocation listener marshallers among the arguments
        int acount = args.length;
        for (int ii = 0; ii < acount; ii++) {
            Object arg = args[ii];
            if (arg instanceof ListenerMarshaller) {
                ListenerMarshaller lm = (ListenerMarshaller)arg;
                lm.requestId = nextRequestId();
                lm.mapStamp = System.currentTimeMillis();
                // create a mapping for this marshaller so that we can properly dispatch responses
                // sent to it
                _listeners.put(lm.requestId, lm);
            }
        }

        // create an invocation request event
        InvocationRequestEvent event = new InvocationRequestEvent(invOid, invCode, methodId, args);
        event.setTransport(transport);

        // because invocation directors are used on the server, we set the source oid here so that
        // invocation requests are properly attributed to the right client object when created by
        // server-side entities only sort of pretending to be a client
        event.setSourceOid(_clobj.getOid());

//         Log.info("Sending invreq " + event + ".");

        // now dispatch the event
        _omgr.postEvent(event);
    }
769170
void sendNextProtocol(NextProtoNego.Provider provider) throws IOException
    {
        if (isInitialHandshake && provider != null)
        {
            if (protocols == null)
            {
                if (NextProtoNego.debug)
                    System.err.println(new StringBuilder("[C] NPN not received for ").append(conn != null ? conn : engine));
                ((NextProtoNego.ClientProvider)provider).unsupported();
            }
            else
            {
                try
                {
                    String protocol = ((NextProtoNego.ClientProvider)provider).selectProtocol(protocols);
                    if (NextProtoNego.debug)
                        System.err.println(new StringBuilder("[C] NPN selected '").append(protocol).append("' for ").append(conn != null ? conn : engine));
                    new NextProtocolMessage(protocol).write(output);
                    output.flush();
                }
                catch(Throwable t)
                {
                    fatalSE(Alerts.alert_handshake_failure, "NPN protocol selection failure", t);
                }
            }
        }
    }