public INDArray getArray ( ) { //start a subscriber that can send us ndarrays if ( subscriber == null ) { running = new AtomicBoolean ( true ) ; subscriber = AeronNDArraySubscriber . startSubscriber ( aeron , subscriberHost , subscriberPort , this , subscriberStream , running ) ; log . debug ( "Started parameter server client on " + subscriber . connectionUrl ( ) ) ; } if ( arr == null ) arr = new AtomicReference <> ( none ) ; log . debug ( "Parameter server client retrieving url from " + ndarrayRetrieveUrl ) ; //note here that this is the "master url" String [ ] split = ndarrayRetrieveUrl . split ( ":" ) ; //The response daemon is always the master daemon's port + 1 //A "master daemon" is one that holds both the //parameter averaging daemon AND the response daemon for being able to send //the "current state ndarray" int port = Integer . parseInt ( split [ 1 ] ) ; int streamToPublish = Integer . parseInt ( split [ 2 ] ) ; //the channel here is the master node host with the port + 1 //pointing at the response node where we can request ndarrays to be sent to //the listening daemon String channel = AeronUtil . aeronChannel ( split [ 0 ] , port ) ; //publish the address of our subscriber //note here that we send the ndarray send url, because the //master also hosts try ( HostPortPublisher hostPortPublisher = HostPortPublisher . builder ( ) . channel ( channel ) . aeron ( aeron ) //note here that we send our subscriber's listening information . streamId ( streamToPublish ) . uriToSend ( AeronConnectionInformation . of ( subscriberHost , subscriberPort , subscriberStream ) . toString ( ) ) . build ( ) ) { hostPortPublisher . send ( ) ; log . debug ( "Sent subscriber information " + AeronConnectionInformation . of ( subscriberHost , subscriberPort , subscriberStream ) . toString ( ) ) ; //wait for array to be available while ( arr . get ( ) == none ) { Thread . sleep ( 1000 ) ; log . info ( "Waiting on array to be updated." ) ; } } catch ( Exception e ) { log . error ( "Error with publishing" , e ) ; } INDArray arr2 = arr . get ( ) ; arr . set ( none ) ; return arr2 ; }
Get an ndarray from the designated ndarray retrieve url . This will pull the current ndarray from the master