title: Servlet-3 Async Context, how to do asynchronous writes?
body: <h2>Problem Description</h2>
  
  <p>Servlet-3.0 API allows to detach a request/response context and answer to it later.</p>
  
  <p>However if I try to write a big amount of data, something like:</p>
  
  <pre><code>AsyncContext ac = getWaitingContext() ;
  ServletOutputStream out = ac.getResponse().getOutputStream();
  out.print(some_big_data);
  out.flush()
  </code></pre>
  
  <p>It may actually block - and it does block in trivial test cases - for both Tomcat 7 and Jetty 8. The tutorials recommend to create a thread pool that would
  handle such a setup - witch is generally the counter-positive to a traditional 10K architecture.</p>
  
  <p>However if I have 10,000 open connections and a thread pool of let's say 10 threads, 
  it is enough for even 1% of clients that have low speed connections or just blocked
  connection to block the thread pool and completely block the comet response or
  slow it down significantly.</p>
  
  <p>The expected practice is to get "write-ready" notification or I/O completion notification 
  and than continue to push the data.</p>
  
  <p>How can this be done using Servlet-3.0 API, i.e. how do I get either:</p>
  
  <ul>
  <li>Asynchronous Completion notification on I/O operation.</li>
  <li>Get non-blocking I/O with write ready notification.</li>
  </ul>
  
  <p>If this is not supported by the Servlet-3.0 API, are there any Web Server specific APIs (like Jetty Continuation or Tomcat CometEvent) that allow to handle such events truly asynchronously without faking asynchronous I/O using thread pool.</p>
  
  <p>Does anybody know?</p>
  
  <p>And if this is not possible can you confirm it with a reference to documentation?</p>
  
  <h2>Problem demonstration in a sample code</h2>
  
  <p>I had attached the code below that emulates event-stream.</p>
  
  <p>Notes:</p>
  
  <ul>
  <li>it uses <code>ServletOutputStream</code> that throws <code>IOException</code> to detect disconnected clients</li>
  <li>it sends <code>keep-alive</code> messages to make sure clients are still there</li>
  <li>I created a thread pool to "emulate" asynchronous operations.</li>
  </ul>
  
  <p>In such an example I explicitly defined thread pool of size 1 to show the problem:</p>
  
  <ul>
  <li>Start an application</li>
  <li>Run from two terminals <code>curl http://localhost:8080/path/to/app</code> (twice)</li>
  <li>Now send the data with <code>curd -d m=message http://localhost:8080/path/to/app</code></li>
  <li>Both clients received the data</li>
  <li>Now suspend one of the clients (Ctrl+Z) and send the message once again <code>curd -d m=message http://localhost:8080/path/to/app</code></li>
  <li>Observe that another non-suspended client either received nothing or after the message was transfered stopped receiving keep-alive requests because other thread is blocked.</li>
  </ul>
  
  <p>I want to solve such a problem without using thread pool, because with 1000-5000 open
  connections I can exhaust the thread pool very fast.</p>
  
  <p>The sample code below.</p>
  
  <hr>
  
  <pre><code>import java.io.IOException;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.LinkedBlockingQueue;
  
  import javax.servlet.AsyncContext;
  import javax.servlet.ServletConfig;
  import javax.servlet.ServletException;
  import javax.servlet.annotation.WebServlet;
  import javax.servlet.http.HttpServlet;
  import javax.servlet.http.HttpServletRequest;
  import javax.servlet.http.HttpServletResponse;
  import javax.servlet.ServletOutputStream;
  
  
  @WebServlet(urlPatterns = "", asyncSupported = true)
  public class HugeStreamWithThreads extends HttpServlet {
  
      private long id = 0;
      private String message = "";
      private final ThreadPoolExecutor pool = 
          new ThreadPoolExecutor(1, 1, 50000L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue&lt;Runnable&gt;());
          // it is explicitly small for demonstration purpose
  
      private final Thread timer = new Thread(new Runnable() {
          public void run()
          {
              try {
                  while(true) {
                      Thread.sleep(1000);
                      sendKeepAlive();
                  }
              }
              catch(InterruptedException e) {
                  // exit
              }
          }
      });
  
  
      class RunJob implements Runnable {
          volatile long lastUpdate = System.nanoTime();
          long id = 0;
          AsyncContext ac;
          RunJob(AsyncContext ac) 
          {
              this.ac = ac;
          }
          public void keepAlive()
          {
              if(System.nanoTime() - lastUpdate &gt; 1000000000L)
                  pool.submit(this);
          }
          String formatMessage(String msg)
          {
              StringBuilder sb = new StringBuilder();
              sb.append("id");
              sb.append(id);
              for(int i=0;i&lt;100000;i++) {
                  sb.append("data:");
                  sb.append(msg);
                  sb.append("\n");
              }
              sb.append("\n");
              return sb.toString();
          }
          public void run()
          {
              String message = null;
              synchronized(HugeStreamWithThreads.this) {
                  if(this.id != HugeStreamWithThreads.this.id) {
                      this.id = HugeStreamWithThreads.this.id;
                      message = HugeStreamWithThreads.this.message;
                  }
              }
              if(message == null)
                  message = ":keep-alive\n\n";
              else
                  message = formatMessage(message);
  
              if(!sendMessage(message))
                  return;
  
              boolean once_again = false;
              synchronized(HugeStreamWithThreads.this) {
                  if(this.id != HugeStreamWithThreads.this.id)
                      once_again = true;
              }
              if(once_again)
                  pool.submit(this);
  
          }
          boolean sendMessage(String message) 
          {
              try {
                  ServletOutputStream out = ac.getResponse().getOutputStream();
                  out.print(message);
                  out.flush();
                  lastUpdate = System.nanoTime();
                  return true;
              }
              catch(IOException e) {
                  ac.complete();
                  removeContext(this);
                  return false;
              }
          }
      };
  
      private HashSet&lt;RunJob&gt; asyncContexts = new HashSet&lt;RunJob&gt;();
  
      @Override
      public void init(ServletConfig config) throws ServletException
      {
          super.init(config);
          timer.start();
      }
      @Override
      public void destroy()
      {
          for(;;){
              try {
                  timer.interrupt();
                  timer.join();
                  break;
              }
              catch(InterruptedException e) {
                  continue;
              }
          }
          pool.shutdown();
          super.destroy();
      }
  
  
      protected synchronized void removeContext(RunJob ac)
      {
          asyncContexts.remove(ac);
      }
  
      // GET method is used to establish a stream connection
      @Override
      protected synchronized void doGet(HttpServletRequest request, HttpServletResponse response)
              throws ServletException, IOException {
  
          // Content-Type header
          response.setContentType("text/event-stream");
          response.setCharacterEncoding("utf-8");
  
          // Access-Control-Allow-Origin header
          response.setHeader("Access-Control-Allow-Origin", "*");
  
          final AsyncContext ac = request.startAsync();
  
          ac.setTimeout(0);
          RunJob job = new RunJob(ac);
          asyncContexts.add(job);
          if(id!=0) {
              pool.submit(job);
          }
      }
  
      private synchronized void sendKeepAlive()
      {
          for(RunJob job : asyncContexts) {
              job.keepAlive();
          }
      }
  
      // POST method is used to communicate with the server
      @Override
      protected synchronized void doPost(HttpServletRequest request, HttpServletResponse response)
              throws ServletException, IOException 
      {
          request.setCharacterEncoding("utf-8");
          id++;
          message = request.getParameter("m");        
          for(RunJob job : asyncContexts) {
              pool.submit(job);
          }
      }
  
  
  }
  </code></pre>
  
  <p>The sample above uses threads to prevent blocking... However if the number of blocking clients is bigger than the size of the thread pool it would block.</p>
  
  <p>How could it be implemented without blocking?</p>