Hello,
I'm developing a java web application for finances quotes' real-time
stream, for hundreds of concurrent users. The solution involves a long
http request serving javascript snippets that will be executed by broser
when received, changing the data shown to the user.
To do this, I have a Comet servlet, ServletStream, that receives the
request, and registers a new instance of JavascriptNotificationListener,
a wrapper to the response.writer, to a running worker thread,
MarketNotifier, that, when a market event occurs, fires the
JavascriptNotificationListener that writes the corresponding event
javascript snippet to the response.writer. There is the Flusher, a
running thread that awakes every half second and do a flush on every
registered response.writer through the same
JavascriptNotificationListener. And there is a servlet, StreamAction,
that submits commands to the MarketNotifier, to change the market event
subscriptions. Some of these commands can also fire market events on the
JavascriptNotificationListener, and are run while the ServletStream is
sending data.
Every IO on the writer (print, println, flush, checkError) is
synchronized on a mutex Object holded by each instance of
JavascriptNotificationListener, that is, I have a mutex for each
response.writer.
When running this solution, I often use to get the following error:
java.lang.NullPointerException
at
org.apache.coyote.http11.InternalNioOutputBuffer.addToBB(InternalNioOutputBuffer.java:607)
at
org.apache.coyote.http11.InternalNioOutputBuffer.commit(InternalNioOutputBuffer.java:600)
at
org.apache.coyote.http11.Http11NioProcessor.action(Http11NioProcessor.java:1025)
at org.apache.coyote.Response.action(Response.java:183)
at org.apache.coyote.Response.sendHeaders(Response.java:379)
at
org.apache.catalina.connector.OutputBuffer.doFlush(OutputBuffer.java:305)
at
org.apache.catalina.connector.OutputBuffer.flush(OutputBuffer.java:288)
at
org.apache.catalina.connector.CoyoteWriter.flush(CoyoteWriter.java:95)
at
br.com.cedro.stream.JavascriptNotificationListener.syn(JavascriptNotificationListener.java:106)
at br.com.cedro.stream.Flusher.run(Flusher.java:42)
This error occurs even on low usage scenarios (five concurrent users)
almost every hour. After the first occurrence happens, every other IO on
all response.writers catches the same exception.
I'm using apache-tomcat-6.0.14, with default Nio Http Connector
configuration parameters on server.xml.
I think this problem has to do with synchronization for IO operations.
Am I right? Or must I simply check some status before doing IO?
The Chat Comet tomcat example does synchronization on a single mutex for
all response.writers. Must I do the same? Why? Or can I use
higher-granularity mutexes like I did? If so, what is the right object
to synchronize? Response? Response.writer?
Has anyone developed Comet servlets and faced the same problem?
Here follow the relevant pieces of my source codes (I put some comments
about the problem to increase readability):
public class JavascriptNotificationListener ... {
// This is the only class that manipulates response.writer
...
private PrintWriter out; // set as response.writer on constructor
private Object writeMonitor = new Object(); // the mutex
public JavascriptNotificationListener(PrintWriter responseWriter){
this.out = responseWriter;
synchronized (writeMonitor) { // Althought at this time we have
just one thread here
out.print("<html><head><script language=\"javascript\">");
}
}
...
public void flush(){
// Called by the Flusher thread every 500 msec
synchronized (writeMonitor) {
out.print("</script>\n<script language=\"javascript\">");
// checkError internally calls flush
if(out.checkError()){
log.error("Error flushing event stream.");
}
}
}
...
public void syn(){
// Called by the Flusher thread every five seconds
synchronized (writeMonitor) {
out.print("s();");
if(out.checkError()){
log.error("Error flushing event stream.");
}
}
...
public void fire(MarketEvent e){
// Called by the worker thread
synchronized (writeMonitor) {
out.print(translate(e));
}
}
...
}
public class ServletStream ... {
// The comet stream servlet
...
private static final Flusher flusher = new Flusher(500); // Will be
started only once on Servlet.init
public void event(CometEvent event) throws IOException,
ServletException {
...
final HttpServletResponse response = event.getHttpServletResponse();
final HttpServletResponse response = event.getHttpServletResponse();
// A user can only have one session on our system
String username = (String)
request.getSession().getAttribute("username");
if(event.getEventType() == CometEvent.EventType.BEGIN){
if(!AuthStore.checkKey((String)
request.getSession().getAttribute("key"))){
// User unauthenticated
response.getWriter().println("...Permission Denied...");
// just to simplify: I write an HTML instead
event.close();
} else {
response.setCharacterEncoding("UTF-8");
// This notify object is active and has a working thread
MarketNotifier notify =
NotifierManager.getOrCreateMarketNotifierForUser(pool, username);
// We wrap our writer inside an object that understands
javascript serialization
JavascriptNotificationListener jsNot = new
JavascriptNotificationListener(response.getWriter());
// Now we subscribe to receive the fire(MarketEvent)
notify.addNotificationListener(jsNot);
// And subscribe to be called on flush() and syn()
flusher.addFlushable(username, jsNot);
// LoadSubscribedEvents is a command to define the
initial listeners for specific market events
// Note that MarketNotifier.process(Command) can call
JavascriptNotificationListener.fire(MarketEvent),
// that is, here we can have another thread acessing
response.writer
// This method is also called by an action servlet (see
the following class)
notify.process(new LoadSubscribedEvents());
}
} else if (event.getEventType() == CometEvent.EventType.ERROR ||
event.getEventType() == CometEvent.EventType.END) {
// Stops receiving syn() and flush()
flusher.removeFlushable(username);
MarketNotifier notify =
NotifierManager.getOrCreateMarketNotifierForUser(pool, username);
notify.process(new UnLoadSubscribedEvents());
NotifierManager.freeMarketNotifierForUser(username + ":" +
sessionId);
event.close();
}
}
}
public class Flusher extends Thread {
// Our Flusher thread
private int msec; // time interval to flush, set on constructor
...
public void run(){
int multiplier = INTERVAL_TO_SYN_IN_SECS * 1000 / msec;
int i = 0;
while(running.get()){
try {
startFlush = new Date();
synchronized (flushMonitor) { // This mutex is also used
on addFlushable and removeFlushable
++i;
for(Flushable f : flushables){
if(i % multiplier == 0){
f.syn();
} else {
f.flush();
}
}
}
try {
long timeToSleep = startFlush.getTime() + msec -
(new Date()).getTime();
if(timeToSleep > 0){
Thread.sleep(timeToSleep);
} else {
log.info("...Flush contention..."); // More than
that, indeed :)
}
} catch (InterruptedException e) {
// Ignore!
}
} catch (Throwable t) {
// Here we catch that exception!
log.error("Fail to flush.", t);
}
}
}
}
public class StreamAction ... {
// Our subscription command processor servlet, called while the
stream is served to change its content
...
public void service(HttpServletRequest request, HttpServletResponse
response) throws ServletException, IOException {
String username = (String)
request.getSession().getAttribute("username");
MarketNotifier notify =
NotifierManager.getOrCreateMarketNotifierForUser(pool, username);
... // Discovers the command to be sent from request.parameters
if(commandName.equals("SubscribeQuoteEvents")){ // Simplified to
easy understanding
// This can call
JavascriptNotificationListener.fire(MarketEvent)
notify.process(new SubscribeQuoteEvents(commandParameters));
}
...
}
}
Any ideas?
Thanks
Leonardo Fraga
Web Developer
Cedro Finances
