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
------------------------------------------------------------------------
No virus found in this incoming message.
Checked by AVG Free Edition.
Version: 7.5.503 / Virus Database: 269.16.2/1142 - Release Date:
11/20/2007 5:44 PM
