Async message processing logic

[viacheslav-fomin-main]

Fixes https://app.clubhouse.io/vgs/story/12753/refactor-littleproxy

Problem:

1. Everything is executed in worker threads
2. A channel is registered with a worker thread not based on worker availability but using round robin algorithm. The thread is selected here https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/common/src/main/java/io/netty/util/concurrent/MultithreadEventExecutorGroup.java#L137. Chooser is https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/common/src/main/java/io/netty/util/concurrent/DefaultEventExecutorChooserFactory.java
3. If one channel blocks a worker, other channel can subscribe to the same thread and get blocked as well. Does not matter how many worker threads there are.

Solutions:

1. Hack main even loop
2. Provide handlers with custom event loops
3. Extract processing logic and run it in a separate executor service

Chosen solution: 3

Comments:
In Netty all the tasks related to the same channel must be executed strictly in order. Event loop executes tasks which are any possible work done by Netty. For example executing fireChannelRead from a separate thread will create a task and put it into main event loop task queue.

Also event loop is shared between channels and all the tasks (for different channels) are in one queue and executed in one loop iteration.

Hacking event loop by adding some sort of cached executor makes LP unstable in some cases. A task may be executed in incorrect order leading to issues which are hard to debug.

And the main issue with that is that it goes opposite to Netty principles.

[viacheslav-fomin-main]

will be provided from outside, possibly in a wrapper with added metrics

[osklyarenko]

i like the idea of a specialized wrapper.

[viacheslav-fomin-main]

initial the channel is assigned on channelRegistered which happens a bit later but we need the channel right now to create global state wrapper event loop

[viacheslav-fomin-main]

this handler contains authentication and payload transformation logic which is executed in a separate executor service

[osklyarenko]

could we move it to a separate class / java file?

[viacheslav-fomin-main]

this class encapsulate some code of client to proxy connection.. it is easier to have it here..

[viacheslav-fomin-main]

we can not do async call when chunks are used. It is due to https://github.com/netty/netty/blob/6fdd7fcddbe964b2f30d7492a926f4f0bf0f083f/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L321.

The app does not use chunks as they are aggregated by https://github.com/verygoodsecurity/LittleProxy/blob/vgs-edition/src/main/java/org/littleshoot/proxy/impl/ClientToProxyConnection.java#L817

[viacheslav-fomin-main]

this event loop will not always be used. Only when a call to fireChannel.. will be executed from another thread (when message processing thread was used). The main context is still handler by InboundGlobalStateHandler .

The reason is that in the same loop fireChannel.. calls are executed like recursion and this recursion starts before we have data in the context (from the request) so GlobalStateWrapperEvenLoop is not executed

[viacheslav-fomin-main]

this is very simple wrapper without any modification to the logic of the main event loop. Any attempt to do that made the proxy unstable

[viacheslav-fomin-main]

this handler uses SimpleChannelInboundHandler which automatically releases the msg. I am still testing that

[osklyarenko]

you must restore the interruption flag here. missing to do so will have have an adverse effect on the further shutdown code below

        try {
            payloadProcessingExecutor.awaitTermination(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.warn("Failed to shutdown payload processing executor properly", e);
        }

        for (EventLoopGroup group : allEventLoopGroups) {
            if (graceful) {
                group.shutdownGracefully();
            } else {
                group.shutdownGracefully(0, 0, TimeUnit.SECONDS);
            }
        }

        if (graceful) {
            for (EventLoopGroup group : allEventLoopGroups) {
                try {
                    group.awaitTermination(60, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();

                    log.warn("Interrupted while shutting down event loop");
                }
            }
        }

[osklyarenko]

also, this shutdown sequence violates the recommended approach, as described in https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

 void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); // Disable new tasks from being submitted
   try {
     // Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); // Cancel currently executing tasks
       // Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     // (Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     // Preserve interrupt status
     Thread.currentThread().interrupt();
   }
 }

[osklyarenko]

Fixed.

[osklyarenko]

why did we change HttpRequest httpRequest to Object httpRequestObj ?

[viacheslav-fomin-main]

yeah, I had some weird behavior with generic here..

[osklyarenko]

Great job 👍