Multithreading in wxPython

This innocent code below looks okay, but it has one major bug here. The bug can cause the application to crash. This is because we try to perform some GUI related stuff in another thread (MyThread). Many GUI toolkits are single-threaded. In other words, the main loop (the loop that starts the GUI application) should only be responsible for handling any GUI related stuff. Any other long running tasks can be done in another thread.

#!/usr/bin/env python

import wx, threading

class MyFrame(wx.Frame):
    def __init__(self):
        wx.Frame.__init__(self, None, -1, "MyApp")
        panel = wx.Panel(self)
        
        self.btn = wx.Button(panel, label="Click Me!")
        self.btn.Bind(wx.EVT_BUTTON, self._do_something)
        
    def _do_something(self, evt):
        MyThread(self).start()
        
class MyThread(threading.Thread):
    def __init__(self, frame):
        threading.Thread.__init__(self)
        self.frame = frame
        
    def run(self):
        wx.MessageDialog(self.frame, message="Test", caption="Test",
                         style=wx.ICON_ERROR | wx.CENTRE).ShowModal()
                             
if __name__ == "__main__":
    app = wx.PySimpleApp()
    frame = MyFrame()
    frame.Centre()
    frame.Show(True)
    app.MainLoop()

To fix this bug, wxPython provides some thread-safe methods, such as wx.CallAfter, wx.CallLater, and wx.PostEvent. A combination of Publisher/Subscribe and wx.CallAfter can eliminate the bug as shown below.

#!/usr/bin/env python

import wx, threading
from wx.lib.pubsub import Publisher

class MyFrame(wx.Frame):
    def __init__(self):
        wx.Frame.__init__(self, None, -1, "MyApp")
        panel = wx.Panel(self)
        
        self.btn = wx.Button(panel, label="Click Me!")
        self.btn.Bind(wx.EVT_BUTTON, self._do_something)
        
        Publisher().subscribe(self._update, "update")
        
    def _do_something(self, evt):
        MyThread().start()
    
    def _update(self, msg):
        # msg.data is the data that was sent in the CallAfter
        wx.MessageDialog(self, message=msg.data, caption="Test",
                         style=wx.ICON_ERROR | wx.CENTRE).ShowModal()
        
class MyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        
    def run(self):
        msg = "Test Message"
        wx.CallAfter(Publisher().sendMessage, "update", msg)
                             
if __name__ == "__main__":
    app = wx.PySimpleApp()
    frame = MyFrame()
    frame.Centre()
    frame.Show(True)
    app.MainLoop()

How to Create a JAR with Dependencies in Gradle

Creating a single executable JAR/fat JAR can be pretty useful. We can do something like this.

java -jar foobar.jar

In Maven, we need to write a pretty verbose XML to do that as shown in my previous blog. In Gradle, this task can be easily achieved in just few lines as shown in the example below.

apply plugin: 'java'

sourceCompatibility = 1.6
targetCompatibility = 1.6

defaultTasks = ['clean', 'jar']

dependencies {
   compile 'org.slf4j:slf4j-api:1.6.2'
   compile 'ch.qos.logback:logback-core:0.9.30'
   compile 'ch.qos.logback:logback-classic:0.9.30'
   compile 'javax.servlet:servlet-api:2.5'
   compile 'org.eclipse.jetty.aggregate:jetty-all:7.5.4.v20111024' 
   
   testCompile 'junit:junit:4.+'
   testCompile 'org.mockito:mockito-core:1.8.5'
}

jar {
   from { configurations.compile.collect { it.isDirectory() ? it : zipTree(it) } }
   manifest { attributes 'Main-Class': 'test.Main' }
}

How to Set Name in the Threads Created by ExecutorService

If we create a java.util.concurrent.ExecutorService without any argument, it will use DefaultThreadFactory, which has a pre-defined thread name, e.g. pool-n-thread-. To be able to set a different name for threads created by the ExecutorService, we need to create our own ThreadFactory like the example below.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class MyThreadFactory implements ThreadFactory {

    private final String name;
    private final AtomicInteger integer = new AtomicInteger(1);
    
    public MyThreadFactory(String name) {
        this.name = name;
    }
    
    /** 
     * {@inheritDoc}
     */
    @Override
    public Thread newThread(Runnable r) {
        return new Thread(r, name + integer.getAndIncrement());
    }
    
    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(5, 
            new MyThreadFactory("Test"));
        try {
            for (int i = 1; i <= 5; i++) {
                es.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            System.out.println(String.format("[%s] Sleeping for 3 secs",
                                Thread.currentThread().getName()));
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
        finally {
            es.shutdown();
            try {
                es.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

How to Schedule Recurring Tasks in Java

Prior to Java 5, to schedule recurring tasks, we need to use java.util.Timer and java.util.TimerTask.

import java.util.Timer;
import java.util.TimerTask;

public class WithTimerTask {
    
    public static void main(String[] args) {
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                System.out.println("Do something");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        Timer timer = new Timer();
        timer.schedule(task, 0, 1000);
    }
}

From Java 5 onwards, we can easily replace java.util.Timer and java.util.TimerTask with java.util.concurrent.ScheduledExecutorService.

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class WithScheduledExecutorService {

    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(1);
        ses.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println("Do something");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, 0, 1000, TimeUnit.MILLISECONDS);
    }
}

java.util.concurrent.ScheduledExecutorService has more advantages as compared to java.util.TimerTask, such as being able to use a thread pool and not being sensitive to system clock. So whenever possible, use java.util.ScheduledExecutorService as opposed to java.util.TimerTask.

How to Get a Return Value from a Thread in Java

If we need to get a return value from a thread in Java, prior to Java 5, we can do something like this.

public class ReturnValueWithThread {
    public static void main(String[] args) {
        MyThread t = new MyThread();
        t.start();

        // busy doing something.....

        while (!t.isDone()) {
            Thread.yield();
        }
        System.out.println(t.getMessage());
    }
    
    static class MyThread extends Thread {
        private boolean done;
        private String message;
        
        /** 
         * {@inheritDoc}
         */
        @Override
        public void run() {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            done = true;
            message = "Hello World";
        }
        
        public boolean isDone() {
            return done;
        }
        
        public String getMessage() {
            return message;
        }
    }
}

In Java 5 onwards, this task can be easily achieved by using java.util.concurrent.ExecutorService and java.util.concurrent.Future.

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ReturnValueWithExecutor {
    public static void main(String[] args) {
        ExecutorService e = Executors.newFixedThreadPool(1);
        Future<String> f = e.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "Hello World";
            }
        });

        // busy doing something...

        try {
            String message = f.get();
            System.out.println(message);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        } catch (ExecutionException e1) {
            e1.printStackTrace();
        }
    }
} 

How to Start Threads at the Same Time in Java

Suppose we need to start few threads in Java at the same time, the easy way to do it by using something like below.

public class WithForLoop {

    private static final int NUM_THREADS = 5;

    public static void main(String[] args) {
        for (int i = 0; i < NUM_THREADS; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(String.format(
                        "%s: Do something....",
                        Thread.currentThread().getName()));
                }
            }).start();
        }
    }
}

The problem with it is that we give a head-start for thread 1 to start first. For most cases, it may not matter, but if let's say we need to make sure all the threads get started at the same time, we can make use of java.util.concurrent.CountDownLatch.

import java.util.concurrent.CountDownLatch;

public class WithCountDownLatch {

    private static final int NUM_THREADS = 5;

    public static void main(String[] args) {
        final CountDownLatch startGate = new CountDownLatch(1);
        final CountDownLatch endGate = new CountDownLatch(NUM_THREADS);
        for (int i = 0; i < NUM_THREADS; i++) {
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        try {
                            startGate.await();
                            System.out.println(String.format(
                                 "%s: Do something....",
                                 Thread.currentThread().getName()));
                        } finally {
                            endGate.countDown();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            t.start();
        }
        
        // this will cause all the threads to get executed concurrently
        startGate.countDown();
        try {
            endGate.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

How to Implement a Shutdown Method in SocketServer.TCPServer for Python 2.5

Prior to Python 2.6, the SocketServer.TCPServer doesn't have a shutdown method. The SocketServer.TCPServer.serve_forever() also blocks. Here is a TCPServer implementation that works in Python 2.5, implements a shutdown method, and support non-blocking request. The implementation here is based on the TCPServer implementation in Python 2.6.

class Py25TCPServer(SocketServer.ThreadingTCPServer):
   def __init__(self, address_tuple, handler):
      SocketServer.ThreadingTCPServer.__init__(self, address_tuple, handler)
      self.__is_shut_down = threading.Event()
      self.__shutdown_request = False

   def serve_forever(self, poll_interval=0.5):
      """Handle one request at a time until shutdown.

         Polls for shutdown every poll_interval seconds. Ignores
         self.timeout. If you need to do periodic tasks, do them in
         another thread.
      """
      self.__is_shut_down.clear()
      try:
         while not self.__shutdown_request:
            r, w, e = select.select([self], [], [], poll_interval)
            if self in r:
               self._handle_request_noblock()
      finally:
         self.__shutdown_request = False
         self.__is_shut_down.set()

   def shutdown(self):
      """Stops the serve_forever loop.

         Blocks until the loop has finished. This must be called while
         serve_forever() is running in another thread, or it will
         deadlock.
      """
      self.__shutdown_request = True
      self.__is_shut_down.wait()

   def _handle_request_noblock(self):
      """Handle one request, without blocking.

         I assume that select.select has returned that the socket is
         readable before this function was called, so there should be
         no risk of blocking in get_request().
      """
      try:
         request, client_address = self.get_request()
      except socket.error:
         return
      if self.verify_request(request, client_address):
         try:
            self.process_request(request, client_address)
         except:
            self.handle_error(request, client_address)
            self.close_request(request)

To make our program work in both Python 2.5 and 2.6+, we can do something like this.

import sys

python_version = sys.version_info
server = None
if python_version[1] < 6:
    server = Py25TCPServer((self.host, self.port), MyHandler)
else:
    server = SocketServer.TCPServer((self.host, self.port), MyHandler)
server.server_forever()