synchronized void callProcedureNT ( final long ciHandle , final AuthUser user , final Connection ccxn , final boolean isAdmin , final boolean ntPriority , final StoredProcedureInvocation task ) { // If paused, stuff a record of the invocation into a queue that gets // drained when un-paused. We're counting on regular upstream backpressure // to prevent this from getting too out of hand. if ( m_paused ) { PendingInvocation pi = new PendingInvocation ( ciHandle , user , ccxn , isAdmin , ntPriority , task ) ; m_pendingInvocations . add ( pi ) ; return ; } String procName = task . getProcName ( ) ; final ProcedureRunnerNTGenerator prntg ; if ( procName . startsWith ( "@" ) ) { prntg = m_sysProcs . get ( procName ) ; } else { prntg = m_procs . get ( procName ) ; } final ProcedureRunnerNT runner ; try { runner = prntg . generateProcedureRunnerNT ( user , ccxn , isAdmin , ciHandle , task . getClientHandle ( ) , task . getBatchTimeout ( ) ) ; } catch ( InstantiationException | IllegalAccessException e1 ) { // I don't expect to hit this, but it's here... // must be done as IRM to CI mailbox for backpressure accounting ClientResponseImpl response = new ClientResponseImpl ( ClientResponseImpl . UNEXPECTED_FAILURE , new VoltTable [ 0 ] , "Could not create running context for " + procName + "." , task . getClientHandle ( ) ) ; InitiateResponseMessage irm = InitiateResponseMessage . messageForNTProcResponse ( ciHandle , ccxn . connectionId ( ) , response ) ; m_mailbox . deliver ( irm ) ; return ; } m_outstanding . put ( runner . m_id , runner ) ; Runnable invocationRunnable = new Runnable ( ) { @ Override public void run ( ) { try { runner . call ( task . getParams ( ) . toArray ( ) ) ; } catch ( Throwable ex ) { ex . printStackTrace ( ) ; throw ex ; } } } ; try { // pick the executor service based on priority // - new (from user) txns get regular one // - sub tasks and sub procs generated by nt procs get // immediate exec service (priority) if ( ntPriority ) { m_priorityExecutorService . submit ( invocationRunnable ) ; } else { m_primaryExecutorService . submit ( invocationRunnable ) ; } } catch ( RejectedExecutionException e ) { handleNTProcEnd ( runner ) ; // I really don't expect this to happen... but it's here. // must be done as IRM to CI mailbox for backpressure accounting ClientResponseImpl response = new ClientResponseImpl ( ClientResponseImpl . UNEXPECTED_FAILURE , new VoltTable [ 0 ] , "Could not submit NT procedure " + procName + " to exec service for ." , task . getClientHandle ( ) ) ; InitiateResponseMessage irm = InitiateResponseMessage . messageForNTProcResponse ( ciHandle , ccxn . connectionId ( ) , response ) ; m_mailbox . deliver ( irm ) ; return ; } }
Invoke an NT procedure asynchronously on one of the exec services .