Re: [jgroups-users] Making outstanding requests using Async methods channel.send()

So you want to implement flow control? I suggest take a look at how this 
is done in JGroups (UFC or MFC).


On 17/02/17 21:13, Questions/problems related to using JGroups wrote:
> <<Locally, each of the 25 invokers is roughly sending 4 (100/25) messages
> &lt;&lt;until it blocks. When the receiver has processed a message, it
> &lt;&lt;decrements the 'credits' (max of 100) to allow for an invoker to
> send
> &lt;&lt;another message.
>
> Yes, this is right. The question is how to make those threads (invoker)
> collaborate so that the total number of outstanding requests (in client
> side) does not exceed 100 requests?
> Sorry my pervious explainion was not clear and please ignore this &quot;On
> other words, I
> want myProtocol always processes 100 requests at the time.&quot;
>
> &lt;&lt;Now you're running 3 instances, so each receiver wil receive a total
> of
> &lt;&lt;25 * 10000 * 3 = 750'000 messages.
> That is true.
>
> &lt;&lt;However, your AtomicInteger is only local, ie. within the local JVM.
>  That is right. I just want each client to have 100 outstanding requests at
> the time, ignoring how many requests myProtocol processes at the time. Again
> I should not have said this &quot;On other words, I want myProtocol always
> processes 100 requests at the time&quot;.
>
> &lt;&lt; I don't know what myProtocol is doing.
> Consensus protocol, ZAB protocol.
>
> &lt;&lt;So what's the question?
> How to make those threads (invoker) collaborate so that the total number of
> outstanding requests does not exceed 100 requests?
>
> I have added synchronized block between receive() method and a thread
> critical section, see code below. But when I add more client instances, the
> clients send few requests and then block (stop sending requests).
>
> public class Clients {
> 	private Credit credits = new Credit(0);
> 	
> 	public void startSend() throws Throwable {
>
> 		// create sender (threads) to send requests
> 		for(int i=0; i &lt; invokers.length; i++){
> 			invokers[i]=new Invoker(channel, requestToSendForEachInv, credits);
> 		}
>
> 		for(Invoker invoker: invokers)
> 			invoker.start();
>
> 		for(Invoker invoker: invokers) {
> 			invoker.join();
> 		}
>
> 	}
>
> 	/*
> 	 * Here I receive a replay. I need to check if the outstanding requests
> 	 * less than 100, if so notify all invokers.
> 	 */
> 	public void receive(Message msg) {
>
> 		synchronized (this) {
> 			synchronized(credits){
> 				if (credits.decAndGet()&lt;100){
> 					credits.notify();
> 				}
> 			}
> 		}
>
> 	}
>
> 	//Sender to send write Request
> 	private class Invoker extends Thread {
>
> 		public void run() {
> 			while(true) {
>
> 				for (int i = 0; i &lt; toSend; i++) {
> 					//What is wrong with below code?
> 					synchronized (credits) {
> 						if(credits.incAndGet()>=100){
> 							credits.wait();
> 						}
> 					}
> 					channel.send(message);
> 				}
>
> 			}
> 		}
>
> 		public static void main(String[] args) {
> 			Clients client = new Clients();
> 			client.startSend();
> 		}
>
> 	}
>
> }
>
>
>
>
> --
> View this message in context: http://jgroups.1086181.n5.nabble.com/Making-outstanding-requests-using-Async-methods-channel-send-tp11287p11289.html
> Sent from the JGroups - General mailing list archive at Nabble.com.
>
> ------------------------------------------------------------------------------
> Check out the vibrant tech community on one of the world's most
> engaging tech sites, SlashDot.org! http://sdm.link/slashdot
> _______________________________________________
> javagroups-users mailing list
> [email protected]
> https://lists.sourceforge.net/lists/listinfo/javagroups-users
>

-- 
Bela Ban, JGroups lead (http://www.jgroups.org)


------------------------------------------------------------------------------
Check out the vibrant tech community on one of the world's most
engaging tech sites, SlashDot.org! http://sdm.link/slashdot