Concurrency on Vertx

Question

i have joined to one of those Vertx lovers , how ever the single threaded main frame may not be working for me , because in my server there might be 50 file download requests at a moment , as a work around i have created this class

public abstract T onRun() throws Exception;
public abstract void onSuccess(T result);
public abstract void onException();

private static final int poolSize = Runtime.getRuntime().availableProcessors();
private static final long maxExecuteTime = 120000;

private static WorkerExecutor mExecutor;

private static final String BG_THREAD_TAG = "BG_THREAD";

protected RoutingContext ctx;

private boolean isThreadInBackground(){
    return Thread.currentThread().getName() != null && Thread.currentThread().getName().equals(BG_THREAD_TAG);
}

//on success will not be called if exception be thrown
public BackgroundExecutor(RoutingContext ctx){

    this.ctx = ctx;

    if(mExecutor == null){
        mExecutor = MyVertxServer.vertx.createSharedWorkerExecutor("my-worker-pool",poolSize,maxExecuteTime);
    }

    if(!isThreadInBackground()){

    /** we are unlocking the lock before res.succeeded , because it might take long and keeps any thread waiting */

    mExecutor.executeBlocking(future -> {
        try{

              Thread.currentThread().setName(BG_THREAD_TAG);

              T result = onRun();
              future.complete(result);

        }catch (Exception e) {
            GUI.display(e);
            e.printStackTrace();
            onException();
            future.fail(e);
        }

              /** false here means they should not be parallel , and will run without order multiple times on same context*/
            },false, res -> {

                if(res.succeeded()){
                    onSuccess((T)res.result());
                }

            });

    }else{


        GUI.display("AVOIDED DUPLICATE BACKGROUND THREADING");
        System.out.println("AVOIDED DUPLICATE BACKGROUND THREADING");

        try{

              T result = onRun();

              onSuccess((T)result);

        }catch (Exception e) {
            GUI.display(e);
            e.printStackTrace();
            onException();          
        }
    }

}

allowing the handlers to extend it and use it like this

public abstract class DefaultFileHandler implements MyHttpHandler{

public abstract File getFile(String suffix);

@Override
public void Handle(RoutingContext ctx, VertxUtils utils, String suffix) {
    new BackgroundExecutor<Void>(ctx) {

        @Override
        public Void onRun() throws Exception {

            File file = getFile(URLDecoder.decode(suffix, "UTF-8"));

            if(file == null || !file.exists()){
                utils.sendResponseAndEnd(ctx.response(),404);
                return null;
            }else{
                utils.sendFile(ctx, file);
            }
            return null;
        }

        @Override
        public void onSuccess(Void result) {}

        @Override
        public void onException() {
            utils.sendResponseAndEnd(ctx.response(),404);
        }
    };
}

and here is how i initialize my vertx server

vertx.deployVerticle(MainDeployment.class.getCanonicalName(),res -> {
          if (res.succeeded()) {

                GUI.display("Deployed");

              } else {
                res.cause().printStackTrace();
              }
            });

    server.requestHandler(router::accept).listen(port);

and here is my MainDeployment class

public class MainDeployment extends AbstractVerticle{

  @Override
  public void start() throws Exception {

    // Different ways of deploying verticles

    // Deploy a verticle and don't wait for it to start

   for(Entry<String, MyHttpHandler> entry : MyVertxServer.map.entrySet()){
       MyVertxServer.router.route(entry.getKey()).handler(new Handler<RoutingContext>() {

            @Override
            public void handle(RoutingContext ctx) {

                String[] handlerID = ctx.request().uri().split(ctx.currentRoute().getPath());

                String suffix = handlerID.length > 1 ? handlerID[1] : null;
                entry.getValue().Handle(ctx, new VertxUtils(), suffix);

            }
        });
   }

  }
}

this is working just fine when and where i need it , but i still wonder if is there any better way to handle concurencies like this on vertx , if so an example would be really appreciated . thanks alot

Answer 1

vertx is a well-designed model so that a concurrency issue does not occur. generally, vertx does not recommend the multi-thread model. (because, handling is not easy.)

If you select multi-thread model, you have to think about shared data..

Simply, if you just only want to split EventLoop Area, first of all, you make sure Check your a number of CPU Cores. and then Set up the count of Instances .

DeploymentOptions options = new DeploymentOptions().setInstances(4);
vertx.deployVerticle("com.mycompany.MyOrderProcessorVerticle", options);

But, If you have 4cores of CPU, you don't set up over 4 instances. If you set up to number four or more, the performance won't improve.

vertx concurrency reference

http://vertx.io/docs/vertx-core/java/

Answer 2

I don't fully understand your problem and reasons for your solution. Why don't you implement one verticle to handle your http uploads and deploy it multiple times? I think that handling 50 concurrent uploads should be a piece of cake for vert.x.

When deploying a verticle using a verticle name, you can specify the number of verticle instances that you want to deploy:

DeploymentOptions options = new DeploymentOptions().setInstances(16); vertx.deployVerticle("com.mycompany.MyOrderProcessorVerticle", options);

This is useful for scaling easily across multiple cores. For example you might have a web-server verticle to deploy and multiple cores on your machine, so you want to deploy multiple instances to take utilise all the cores.

http://vertx.io/docs/vertx-core/java/#_specifying_number_of_verticle_instances