vertx event handling priorities

[Jason Judt]

Hi,

I'm quite new to vertx and asynchronous programming. I have implemented a simple log server that accepts log records (java.util.logging) over network and writes them to files. I implemented a simple rollover ability when the log file gets too big to move the main file to .1, .2, .3, etc. extensions. To get this to work, in my netsocket buffer handler i periodically check if the main log file is full. if so, I execute the rollover. My first attempt resorted to synchronous calls and worked fine. It also seemed anti-vertx so I implemented an async version. Under lights loads the async version works as well, but I wrote a test app to blast the log server with 4 threads sending hundreds of thousands of messages. Under such a load, it seems that vertx will happily execute existing and new buffer events from the socket, and even though during previous operations i'm adding handlers of my own to execute rolloevers, They never run until the socket is not busy. So, rather than rolling over to a new log file when the current log file is 1 MB (for example), my handlers don't run until the client tester finishes sending all its data (40 MB in this example). The entire log data will be written to the main log file. Then, after some delay, all my queued up handlers get invoked later.

I guess my question is, is there a fairness policy to executing handlers by the vertx event thread? Is there a way to prioritize my handlers so they do not get deferred under network load? I didn't see anything mentioned in manuals or apis that indicated such was the case, but I can't explain this behavior I'm experiencing. 

Thanks,

Jason

[Julien Viet]

Hi,

I understand what you want to acheive, however I don’t understand how you are currently proceeding.

could you re-explain again (and not in a single chunk of text please :-) ).

Julien

--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+un...@googlegroups.com.
Visit this group at https://groups.google.com/group/vertx.
To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/a512aa5c-d55c-4ea8-9c52-df145d54dd4e%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Jason Judt]

Yeah, i was surprised by the lack of response, then realized I should maybe post code to better explain my issue. Tonight I'll work on trimming down my messy code to something small that demonstrates my issue and post that. 

The summary is, in a netsocket buffer handler, i create handlers of my own. My own handlers won't get run, though, if the netsocket keeps receiving more buffer events due to high load. I was under the impression that handlers would be queued up and run in a first-in-first-out sort of manner, but what I experience seems to be more of a priority queue where netsocket buffer events out rank my own handlers. 

Thanks for the response. 

Jason 

[Julien Viet]

reactor events are never reordered, so it would be best you explain your current design so we can help you figure out what is going on.

note : you can pause/resume NetSocket to apply back-pressure if you need, for instance you could pause the netsocket while rotating the file and resume after. But perhaps it is not something you want to do according to your constraints.

To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/13fcf6d4-8fc2-45e9-b715-dcd3ead26567%40googlegroups.com.

[Jason Judt]

I think we can disregard my question. I attempted to simplify my code and run it to prove my concern. Unfortunately for me, the code ran as Julien said: the events were never reordered. Somehow in my main app they aren't running in the order I expected but I'll have to spend some time figuring out where I have a bug. For reference here is my simplified code, but it works as expected where the reset handler does get invoked in a timely manner even when the socket buffer handler is under tremendous loads. Thanks for the help. 

package net.home;

import io.vertx.core.AbstractVerticle;

import io.vertx.core.Future;

import io.vertx.core.net.NetServer;

import io.vertx.core.net.NetServerOptions;

import io.vertx.core.net.SocketAddress;

import net.home.vertxLogger.LogRecord;

import net.home.vertxLogger.ParserHandler;

public class AsyncLogger extends AbstractVerticle{

private NetServer socketListener;

private int buffersRecv = 0;

private int bytesRecv = 0;

@Override

public void start() throws Exception {

int port = 11112;

NetServerOptions opts = new NetServerOptions().setPort(port).setHost("localhost");

socketListener = vertx.createNetServer(opts);

// setup the handler

socketListener.connectHandler(socket -> 

{

SocketAddress address = socket.remoteAddress();

System.out.println("connection from: " + address.toString());

socket.handler(buffer -> {

// do something

buffersRecv += 1;

bytesRecv += buffer.length();

if (buffersRecv % 100000 == 0)

{

System.out.println("got 1k buffers, total bytes: " + bytesRecv);

Future<Void> resetFuture = Future.future();

resetFuture.setHandler(rf -> {

System.out.println("reset");

bytesRecv = 0;

});

resetFuture.complete();

}

});

socket.closeHandler( closeHandler -> {

});

});

// tell it to listen

socketListener.listen(res -> {

if (res.succeeded()) {

System.out.println("Server is now listening");

} else {

System.out.println("failed to listen");

}

});

}

}