JBossDeveloper

Ok, I wrote a little byteman script to trace calls to EndpointAssociation.setEndpoint and EndpointAssociation.getEndpoint:

# rules to track calls to getEndpoint and setEndpoint in CXF code
#

RULE trace setEndpoint
CLASS EndpointAssociation
METHOD setEndpoint
IF TRUE
DO trace("foo", "setEndpoint(" + $1.getName() + ")\n" + formatStack())
ENDRULE

RULE trace getEndpoint
CLASS EndpointAssociation
METHOD getEndpoint
AT RETURN
IF TRUE
DO trace("foo", "getEndpoint() ==> " + $!.getName() + "\n" + formatStack())
ENDRULE

When I ran my tests with  this script installed I got a few cases where successive gets and sets were using different endpoints. Here is a dump of the (filtered and reformatted) trace output:

setEndpoint(Interop11ATParticipantService)
Stack trace for thread http-127.0.0.1-8080-3
getEndpoint() ==> Interop11ATParticipantService
Stack trace for thread default-workqueue-1      // n.b. worker 1 from participant executor

setEndpoint(ActivationService)
Stack trace for thread http-127.0.0.1-8080-4
getEndpoint() ==> ActivationService
Stack trace for thread http-127.0.0.1-8080-4

setEndpoint(RegistrationService)
Stack trace for thread http-127.0.0.1-8080-4
getEndpoint() ==> RegistrationService
Stack trace for thread http-127.0.0.1-8080-4

setEndpoint(CompletionCoordinator)
Stack trace for thread http-127.0.0.1-8080-4
getEndpoint() ==> CompletionCoordinator
Stack trace for thread default-workqueue-1

setEndpoint(CompletionInitiator)
Stack trace for thread http-127.0.0.1-8080-5
getEndpoint() ==> CompletionInitiator
Stack trace for thread default-workqueue-2      // n.b. worker 2 from initiator executor

setEndpoint(Interop11ATInitiatorService)
Stack trace for thread http-127.0.0.1-8080-6
getEndpoint() ==> Interop11ATInitiatorService
Stack trace for thread default-workqueue-2

. . .

setEndpoint(RegistrationService)
Stack trace for thread http-127.0.0.1-8080-9
getEndpoint() ==> RegistrationService
Stack trace for thread http-127.0.0.1-8080-9

setEndpoint(RegistrationService)
Stack trace for thread http-127.0.0.1-8080-9
getEndpoint() ==> RegistrationService
Stack trace for thread http-127.0.0.1-8080-9

setEndpoint(Interop11ATInitiatorService)         // hmm, call is to initiator service
Stack trace for thread http-127.0.0.1-8080-9
getEndpoint() ==> Interop11ATParticipantService
Stack trace for thread default-workqueue-1       // oops wrong worker and endpoint!

setEndpoint(ActivationService)
Stack trace for thread http-127.0.0.1-8080-9
getEndpoint() ==> ActivationService
Stack trace for thread http-127.0.0.1-8080-9

setEndpoint(Interop11ATParticipantService)       // hmm, call is participant service
Stack trace for thread http-127.0.0.1-8080-5
getEndpoint() ==> Interop11ATInitiatorService
Stack trace for thread default-workqueue-2       // wrong worker and endpoint again!

setEndpoint(ActivationService)
Stack trace for thread http-127.0.0.1-8080-5
getEndpoint() ==> ActivationService
Stack trace for thread http-127.0.0.1-8080-5

setEndpoint(Interop11ATParticipantService)
Stack trace for thread http-127.0.0.1-8080-8
getEndpoint() ==> Interop11ATParticipantService
Stack trace for thread default-workqueue-3

. . .

So, worker thread 1 was added under a call to the participant service and hence has this endpoint in its inherited therad local. However it appears to get reused to handle the initiator calls later. How so? I added some trace rules to see what exector was being used:

RULE trace getExecutor return value
CLASS ContextUtils
METHOD getExecutor
AT RETURN
IF TRUE
DO traceln("foo", "getExecutor() ==> " + $!)
ENDRULE

RULE trace getExecutor internal
CLASS ContextUtils
METHOD getExecutor
AFTER CALL getNamedWorkQueue
IF TRUE
DO traceln("foo", "getExecutor called getNamedWorkQueue")
ENDRULE

RULE trace getExecutor internal 2
CLASS ContextUtils
METHOD getExecutor
AFTER CALL getAutomaticWorkQueue
IF TRUE
DO traceln("foo", "getExecutor called getAutomaticWorkQueue")
ENDRULE

RULE trace getExecutor internal 3
CLASS ContextUtils
METHOD getExecutor
AFTER CALL getInstance
IF TRUE
DO traceln("foo", "getExecutor called getInstance")
ENDRULE

This first rule prints the return value on the stack ($!) when getExecutor is about to return i.e. at the point where the original handler thread is queueing the runnable to execute the handed off message. The remaining rles show the path followed through the code. Here is the output

setEndpoint(Interop11ATParticipantService)
Stack trace for thread http-127.0.0.1-8080-3
getExecutor called getNamedWorkQueue
getExecutor called getAutomaticWorkQueue
getExecutor() ==> org.apache.cxf.workqueue.AutomaticWorkQueueImpl@ea16c [queue size: 0, max size: 256, threads: 0, active threads: 0, low water mark: 5, high water mark: 25]
Stack trace for thread http-127.0.0.1-8080-3
getEndpoint() ==> Interop11ATParticipantService
Stack trace for thread default-workqueue-1
. . .

setEndpoint(Interop11ATInitiatorService)
Stack trace for thread http-127.0.0.1-8080-5
getExecutor called getNamedWorkQueue
getExecutor called getAutomaticWorkQueue
getExecutor() ==> org.apache.cxf.workqueue.AutomaticWorkQueueImpl@ea16c [queue size: 0, max size: 256, threads: 1, active threads: 0, low water mark: 5, high water mark: 25]
Stack trace for thread http-127.0.0.1-8080-5
getEndpoint() ==> Interop11ATInitiatorService
Stack trace for thread default-workqueue-2
. . .
setEndpoint(Interop11ATParticipantService)
Stack trace for thread http-127.0.0.1-8080-6
getExecutor called getNamedWorkQueue
getExecutor called getAutomaticWorkQueue
getExecutor() ==> org.apache.cxf.workqueue.AutomaticWorkQueueImpl@ea16c [queue size: 0, max size: 256, threads: 2, active threads: 0, low water mark: 5, high water mark: 25]
Stack trace for thread http-127.0.0.1-8080-6
getEndpoint() ==> Interop11ATParticipantService
Stack trace for thread default-workqueue-3
. . .
setEndpoint(Interop11ATParticipantService)
Stack trace for thread http-127.0.0.1-8080-6
getExecutor called getNamedWorkQueue
getExecutor called getAutomaticWorkQueue
getExecutor() ==> org.apache.cxf.workqueue.AutomaticWorkQueueImpl@ea16c [queue size: 0, max size: 256, threads: 5, active threads: 0, low water mark: 5, high water mark: 25]
Stack trace for thread http-127.0.0.1-8080-6
getEndpoint() ==> Interop11ATInitiatorService
Stack trace for thread default-workqueue-2
. . .

As you can see the code fetrches an automatic qork queue by calling getAutomaticWorkQueue and this is clearly allocated once per service rather than onced per endpoint. the result is that workpool threads inherit one endpoint when they are added to the pool but when they are reused they are given a message which is associated with a different endpoint.

Alessio Soldano wrote:

 

Andrew, thanks for the great analysis. In few words, it seems to me that what your tests reveal is that the threadlocal solution for getting the jbossws-spi endpoint in the jbossws custom invoker cannot work in the asynch dispatch happening in usecases like yours.

 

yes, that pretty much sums it up.

Alessio Soldano wrote:

 

I'm thinking of an alternative solution; it should be possible to use the Exchange associated to a given message as the mean for shipping the endpoint instance to the invoker. I'm thinking of using a custom ServletDestination for adding the endpoint to the exchange after it's been created. I'll try this later today or early tomorrow morning, then I'll most probably need you to help again with testing the solution.

Thanks for now

A simpler solution might be to store the current endpoint in the runnable created in ContextUtils.rebaseResponse and then restore it when processing continues i.e.

                        final Endpoint ep = EndpointAssociation,.getEndpoint()

                        // pause dispatch on current thread ...
                        inMessage.getInterceptorChain().pause();

 

                         // ... and resume on executor thread
                        getExecutor(inMessage).execute(new Runnable() {
                            Endpoint savedEp = ep;
                            public void run() {
                                EndpointAssociation.setEndpoint(savedEp);
                                inMessage.getInterceptorChain().resume();
                            }
                        });

Of course this also means that EndpointAssociation can be changed to use a conventional ThreadLocal rather than an InheritedThreadLocal to hold the association.

HI Alessio,

Thanks for the quick fix. I'll try to test it as soon as possible but that will probably be tomorrow (Tuesday) rather than today.