Anyvent

anyvent

The pattern always fascinated me. I found it interesting, useful, and compatible with object-oriented programming. A lot of us know it thanks to Node.js. event loop For some reason, recently, I have been looking to implement it and experiment with it. And I finally got the opportunity to do so at work. Event Loop Concepts An event loop doesn’t do too much. We can ask it to start and stop itself. And once started it waits and reacts to events. The events could signal a new message in a queue; an expired timeout and so on. Events trigger an action or a script once they occur. With this in mind, we can already define these interfaces. They’re evoking the aforesaid behaviors: interface EventLoop {
  void start();

  void stop();
}

interface Events {
  Optional<Event> next();
}

interface Event {
  void trigger(Script script);
}

interface Script {
  void run(JsonObject properties, Consumer<Instant> onSuccess, Consumer<Throwable> onFailure);
} The interface deserves more attention. We can ask it to with given properties and with two callbacks. One for success and one for failure. The first one sends the script end time. The second one sends the exception that has occured. Script run Simple event loop The simplest event loop that comes to mind uses : busy waiting final class BusyWaitingEventLoop implements EventLoop {
  private final Events events;
  private final Script script;
  private final AtomicBoolean alive;

  BusyWaitingEventLoop(final Events events, final Script script) {
    this(events, script, new AtomicBoolean(true));
  }
  
  BusyWaitingEventLoop(final Events events, final Script script, final AtomicBoolean alive) {
    this.events = events;
    this.script = script;
    this.alive = alive;
  }

  @Override
  public void start() {
    while (alive.get()) {
      events.next().ifPresent(event -> event.trigger(script));
    }
  }

  @Override
  public void stop() {
    alive.set(false);
  }
} Once started it checks repeatedly for a new event. Once the event is found, it’ll trigger the script according to its own rules. But this approach has a major drawback: the event handling blocks the event loop. This means that we can handle only one event at a time. Multithreaded Event Loop An approach to resolve the aforesaid issue is to start the event loop two or more times. Each time in a different thread. So, here is a potential multithreaded event loop implementation exploiting the : ExecutorService The responsibility is to run tasks. It can run them in different threads or in a single thread. Java provides both implementations thanks to a bunch of . ExecutorService factory methods final class MultithreadedEventLoop implements EventLoop {
  private final EventLoop origin;
  private final Integer nThreads;
  private final ExecutorService executorService;

  MultithreadedEventLoop(final EventLoop origin, final Integer nThreads, final ExecutorService executorService) {
    this.origin = origin;
    this.nThreads = nThreads;
    this.executorService = executorService;
  }

  @Override
  public void start() {
    for (var i = 0; i < nThreads; i++) {
      executorService.execute(origin::start);
    }
  }

  @Override
  public void stop() {
    origin.stop();
    shutdownExecutorService();
  }

  private void shutdownExecutorService() {
    // Java specific code
  }
} We can start a with eight threads in this way: MultithreadedEventLoop public final class Main {
  public static void main(String[] args) {
    var nThreads = 8;
    new MultithreadedEventLoop(
      new BusyWaitingEventLoop(
        anEventsObject,
        aScriptObject
      ),
      nThreads,
      Executors.newFixedThreadPool(nThreads)
    );
  }
} Now we can handle eight events in parallels. An improvement, but we can do it even better once we have resolved the main issue: the event loop and the script running in the same thread. Asynchronous Script The solution to the aforesaid issue is to run scripts and event loops in different threads. In this way, we can have threads dedicated to the scripts and threads dedicated to the event loop. We already have a multithreaded event loop but we are missing the other piece: an asynchronous script. Here is an implementation that exploits the again: ExecutorService final class AsyncScript implements Script {
  private final Script origin;
  private final ExecutorService executorService;

  AsyncScript(final Script origin, final ExecutorService executorService) {
    this.origin = origin;
    this.executorService = executorService;
  }

  @Override
  public void run(final JsonObject properties, final Consumer<Instant> onSuccess, final Consumer<Throwable> onFailure) {
    if (!executorService.isShutdown()) {
      executorService.execute(() -> origin.run(properties, onSuccess, onFailure));
    }
  }
} is a decorator. Using this, we can turn a synchronous to an asynchronous one. AsyncScript Script Script Indeed, we can easily integrate in the previous example: AsyncScript public final class Main {
  public static void main(String[] args) {
    var eventLoopThreads = 2;
    var scriptThreads = 14;
    var executorService = Executors.newFixedThreadPool(eventLoopThreads + scriptThreads);
    new MultithreadedEventLoop(
      new BusyWaitingEventLoop(
        anEventsObject,
        new AsyncScript(
          aScriptObject,
          executorService
        )
      ),
      eventLoopThreads,
      executorService
    ).start();
  }
} Now we have an with sixteen threads. Two threads were assigned to the event loop, with fourteen available to the scripts. So the event loop will receive new events in two threads. While the scripts will run in others fourteen threads. ExecutorService Java Tip A possible improvement specific to Java is to use generics. Something like: interface EventLoop {
  void start();

  void stop();
}

interface Events<I, O> {
  Optional<Event<I, O>> next();
}

interface Event<I, O> {
  void trigger(Script<I, O> script);
}

interface Script<I, O> {
  void run(I in, Consumer<O> onSuccess, Consumer<Throwable> onFailure);
} In this way we can handle generic input/output in the . Script Real-World Scenario While the aggregates objects, in my case, it aggregates multiple that emit . The latter triggers the scripts with the message body as properties. And on completion they remove themselves from the queue. AggregateEvents Events Azure Queues Events AzureQueueMessage Conclusion We can improve the presented implementation in various ways. For example, we can add more capabilities to the event loop, such as restarting. But, as always, keep in mind the modeled domain. Furthermore there could be more opportunities to catch. For example I can guess a prioritised implementation, or I can guess event loops with different implementations. In this way, we could have finer-grained control over threads or priorities. Events ExecutorService Currently, I’m investigating a broader and more consistent way to apply this pattern. So, stay tuned and happy looping to you all!

Implementing an Event Loop in Java for Fun and Profit

Original article available at https://habrahabr.ru/company/mailru/blog/341498/ Tarantool is not just a , it’s an application server with a database on board, which makes it a of choice for quickly and easily implementing certain things people usually spend a lot of time on. database tool I decided to write this article after reading about a RabbitMQ-based user notification service. Quite a lot of IT specialists face similar problems, and I’d like to share with you how I solve them. A few years ago we had a client who wanted to receive real-time (with network-defined latency) notifications about certain events in a web application. Let’s take a look at the requirements: On average, we need to listen for 10 events on the page. There are a million users. Persistence is a must (we don’t want to lose events). Given this incomplete information, let’s make a few assumptions and guesstimates. If this application is a chat with a million users constantly writing something and it takes a single user 10 seconds to create a message, then a million users generate 100,000 messages per second. Here arises the first question: what do we need to deliver? Events or data? This is a classical dilemma of all queues, aggregators, and event servers: what’s a storage unit? So let’s introduce a couple of definitions: is a informational structure that stores information about the of an event. Event minimal fact An event may have some data associated with it: is a full data set, including data weakly relevant to the event. Data Here’s an example: user placed an order to transport cargo from point to point and paid for it with a card and so on. 345 X A B Z Let the information about who (event source) placed an order (fact of an event) constitute an , then our would contain the starting point, the type of cargo, and other information about the user, the order, and the cargo. In other words, an event is a structure that describes the fact of an event, whereas data is all the rest. event data Distinction between an event and data is somewhat arbitrary, but still. Here’s an example of an event: {“type”: “order”,“user”: 345,“status”: “created”,“orderid”: 12345} User placed order . At shipment time, the order had the status (this is superfluous information). 345 12345 created Below I’ll try to formulate some for making the right architectural decision: rule of thumb When designing queues, event servers, or aggregators, servers should handle events, not data. But let me reiterate that: the dividing line between an event and data is blurry: in the example above, an event contains a part of data (the field). status Back to the task: we’re building an and therefore can estimate the traffic. Say, on average, an event is a JSON hash 4 to 10 elements long (that is 60–160 bytes of text). So, given a million users generating 100,000 messages per second, the event flow would be 6 to 16 MB/s. To handle all this traffic, a single-host network with a 200 Mb/s throughput should be enough. event server Now let’s try to calculate how many resources it would take to deliver these events to a million users. Architectures may differ, but there always exist some common principles. If one message needs to be delivered to a million users, it’s highly likely a bug in the architecture (although it might be a legitimate use case). We need to come up with some estimate. Let’s assume that a single message is delivered to 10 users: in a typical messenger, you don’t often have more than 10 friends online; in case of an order, there are rarely more than contractors, and so on. average If we do the math, it takes approximately 2 to 3 Gb of traffic to deliver notifications to our users. Hence another key consideration: this task can be solved on a single server with a 10 Gb network interface controller and about 10 GB of RAM (if we set the caching interval to 10 minutes). Now that we know this system can be implemented on just one server, let’s try to build a real scalable distributed system. Storing data One of the fastest ways of storing cached data on disk is by using a write-ahead log (WAL): data goes to the RAM cache and is then written to the WAL. Since data is always only to the WAL, it’s possible to leverage almost 100% of disk write throughput this way. I won’t discuss the the downsides of WALs here, but let me say they’re well-suited for replication. appended DBMS has just such a WAL: it not only allows replicating data to another host, but also offers two-way asynchronous master-master replication. Tarantool run on a 2012 modest-spec laptop with 220-byte messages yield the performance of 160–180K writes per second, which is almost twice more than our task demands. Tarantool benchmarks Delivering data to a client There are numerous ways of data delivery (we’ll discuss them later). For now, let’s focus on the algorithmic side of the delivery problem. For data delivery to function properly in the real world, we’ll list several requirements it must meet: It must be disconnection-tolerant (messages must not be lost when reconnecting after a disconnect). The client must be thin (most of the logic should ideally be implemented on the server, not on the client). Based on these requirements, through trial and error, we’ve arrived at the following scheme: The client “remembers” (no need for persistence here) the number of the last received message. This number is used when restoring the event server connection. Given this algorithm, we have two candidates for organizing data transport: regular long-polling (each request carries the number of the last received message) or web sockets (the number of the last received message is sent only on reconnect). Data/event scheme After some experimentation, we decided that each event should be uniquely identified by an event key, which, in the general case, is just a string ID of an event. Since event keys are often a combination of several different IDs, let each event key be represented by some array of string IDs. A quick example: user writes a message to chat and mentions user . As a result, we generate an event with a key , which is delivered to all users in chat , and another event with a key , which only user receives. 123 345 567 [ ‘chat’, 345 ] 345 [‘user’, 567] 567 A more detailed view of these events may look like this: {“key”: [ “chat”, 345 ],“data”: {“type”: “new_message”,“msgid”: 9876}} and {“key”: [ “user”, 567 ],“data”: {“type”: “notice”,“chatid”: 345,“msgid”: 9876}} Now we’ve come to how message keys are formed. It doesn’t make much sense (and may even do more harm than good sometimes) to have lots of message keys for more or less similar things. New keys should only be allocated for significantly different entities. : it makes sense to use a single key for these tasks: Example 1 notifying all chat users about a new message; notifying all chat users about a newly connected user. : it makes sense to use different keys for these tasks: Example 2 notifying all chat users about a new message; notifying a user of a chat X about a mention. That is to say, a message key should approximately define message recipients and be treated similarly to an address on an envelope. Implementation Now that we’ve decided on the data/event scheme, it’s time to dive deeper into the implementation details. We’ll store messages in a flat table with the following columns: message ID (ever-incrementing sequence number); time of event generation; message key; message data. We’ll also need two indices for this table: primary index on message ID. This index will be used for deleting old messages from the database; secondary index for selecting messages by key (composite index consisting of a message key and a message ID). The resulting schema is available . here Tarantool DBMS lets us easily create pub/sub applications with a built-in module. fiber Each client request is processed in a separate fiber (a lightweight type of thread). This paradigm makes it a breeze to handle tens of thousands of connections on a single CPU core. That said: your code is not riddled with callbacks (which is the case with Node.js); the gets easily resolved. C10k problem The subscribe algorithm for one key looks like this: Check if there’s any data (new events) associated with a given key. If so, return the data. Add the current fiber to the list of fibers subscribed to the given key. Sleep for some time (so that mobile networks don’t disrupt the functioning of web sockets; 25 seconds was empirically found to work well). Respond to a user (response may be empty, more on that later). The push algorithm is as follows: Write new event to the database. If there are any clients subscribed to the keys that have been added to the database, wake up their corresponding fiber(s). All the server logic, including scaling out to several CPUs/servers, was implemented in less than 500 lines of Lua code. This system, based on three Tarantool instances that run on a single (OpenVZ node) server, utilizes 10% of each of the two allocated CPU and currently serves about 50,000 users. virtual cores If the calculations are correct, one such server is capable of handling around 500,000 users. One or two more CPU cores might be needed, though. Problems associated with the number of sockets per host that are mentioned in the article above are resolved the same way. Removing old messages Each master instance has a background daemon that removes old messages from the database. The algorithm is pretty straightforward: Choose the oldest message (the one with the lowest ID). Check its creation time. If its lifetime hasn’t expired, wait the necessary amount of time. Delete the old message. Get back to step 1. Scaling We started building this system at the time of Tarantool 1.5’s release, which didn’t have two-way asynchronous replication. That’s why its architecture looks like this: Master server (messages can be pushed to it); Replicas (clients can connect to them). Master and replicas are fully identical instances, except that (for now) messages are pushed only to one server. That is to say, the system is currently scaled by adding new replicas, and the peak performance is capped by the performance of the master server (for a server unit, it’s about 400–500K messages per second). modern Update plans Since Tarantool 1.6 has two-way master-master replication, this feature can be used for scaling the system as well. What I have in mind (yet to be implemented): Convert a message ID to an array: master server ID, message ID. Between two reconnects, a client “remembers” this array of values. PROFIT! Other than that, the algorithm stays the same. This way the system can be scaled out to 10–30 master servers (which translates to 4–20 million outgoing messages per second) without making any substantial changes to the architecture. Downsides (there always are some) Lua (greatest downside). It’s simple, but its limitations (1 GB of RAM per instance) force you to scale a bit earlier than you would otherwise have. Unfortunately, we haven’t made public the HTTP part that handles data transport (because it depends on our internal-only modules). It includes a simple asynchronous HTTP server for long polling or a more complex asynchronous server for web sockets + long polling. It takes about 200 lines of code to implement this server in Perl + AnyEvent. Client authorization We don’t use client authorization on the event server (in this context, a client is a web site user), because we don’t think it’s necessary. But, generally speaking, it’s easy to implement: add to each message a key-value pair that defines “who is allowed to see this data” and compare this pair to the information contained in request cookies. Since our server operates on events and not data, we didn’t need to introduce authorization so far. A few words about data transport We started working in this area back when web sockets only started to emerge and the first standards were being drafted. For this reason, regular HTTP long polling has been used for data transport for a very long time. Experimenting with web sockets in mobile networks has shown that: An event (even an empty one) must be passed through a web socket once or twice a minute. Otherwise, mobile networks shut down the connection. There’s no difference between a web socket and long polling + keep-alive in this case. There are lots of mobile devices where browsers don’t support web sockets. That’s why, when talking about interactive web sites in the context of mobile networks, you can say web sockets are gaining traction (supported by 70% of devices), but they still have a long way to go until universal adoption (30% is quite a lot). Applications If a web project includes a queue and an event server, many things that tend to cause problems in other architectures can easily be dealt with. One such thing, and quite obvious at that, is a chat described above. Another elegant application of an event server is working with long-running or resource-intensive algorithms (report generation, statistics gathering, and even video decoding). Suppose we want to decode users’ videos. It goes without saying that it’s a lengthy procedure and not suitable for an HTTP request handler. After uploading a video, a user wants to understand what’s going on. We’ll just enqueue a video decoding task so that the client-side JavaScript starts tracking events associated with it. The video conversion process will send out events as the task gradually nears completion. Based on these events, the user’s browser can display a neat and, what’s more important, accurate progress bar.

Stories, News & Mentions about ANYVENT

Anyvent’s Stories on HackerNoon