Index: ChangeLog
===================================================================
--- ChangeLog	(revision 206)
+++ ChangeLog	(working copy)
@@ -1,6 +1,7 @@
 1.0 "Asieoniezi"
 
 0.126 "TBA"
+ 420618 Decode in a worker thread so the GUI and network don't freeze (Calin Culanu)
  411446 Use `Message-ID' rather than `Message-Id' when posting. (Charles)
  411447 Header pane loses selection when deleting articles (Jeff Berman)
  411657 Post window's cursor appears in the wrong place (Frank Tabor)
Index: po/POTFILES.in
===================================================================
--- po/POTFILES.in	(revision 206)
+++ po/POTFILES.in	(working copy)
@@ -33,6 +33,7 @@
 pan/gui/server-ui.cc
 pan/gui/task-pane.cc
 pan/gui/url.cc
+pan/tasks/decoder.cc
 pan/tasks/nntp.cc
 pan/tasks/nntp-pool.cc
 pan/tasks/socket-impl-gio.cc
Index: pan/usenet-utils/utf8-utils.cc
===================================================================
--- pan/usenet-utils/utf8-utils.cc	(revision 206)
+++ pan/usenet-utils/utf8-utils.cc	(working copy)
@@ -186,7 +186,7 @@
 {
   std::string s = content_to_utf8 (header, fallback_charset1, fallback_charset2);
   if (header.strstr ("=?")) {
-    char * decoded (g_mime_utils_8bit_header_decode ((const guchar*) s.c_str()));
+    char * decoded (g_mime_utils_8bit_header_decode (s.c_str()));
     s = clean_utf8 (decoded);
     g_free (decoded);
   }
Index: pan/gui/pan.cc
===================================================================
--- pan/gui/pan.cc	(revision 206)
+++ pan/gui/pan.cc	(working copy)
@@ -27,6 +27,7 @@
 #include <pan/general/debug.h>
 #include <pan/general/log.h>
 #include <pan/general/file-util.h>
+#include <pan/general/worker-pool.h>
 #include <pan/tasks/socket-impl-gio.h>
 #include <pan/tasks/task-groups.h>
 #include <pan/tasks/nzb.h>
@@ -183,7 +184,6 @@
   bind_textdomain_codeset (GETTEXT_PACKAGE, "UTF-8");
   textdomain (GETTEXT_PACKAGE);
 
-  g_thread_init (0);
   gtk_init (&argc, &argv);
   g_mime_init (GMIME_INIT_FLAG_UTF8);
 
@@ -200,9 +200,10 @@
       url = tok;
     else if (!strcmp(tok,"--no-gui") || !strcmp(tok,"--nogui"))
       gui = false;
-    else if (!strcmp (tok, "--debug"))
-      _debug_flag = true;
-    else if (!strcmp (tok, "--nzb"))
+    else if (!strcmp (tok, "--debug")) { // do --debug --debug for verbose debug
+      if (_debug_flag) _debug_verbose_flag = true;
+      else _debug_flag = true;
+    } else if (!strcmp (tok, "--nzb"))
       nzb = true;
     else if (!strcmp (tok, "--version"))
       { std::cerr << "Pan " << VERSION << '\n'; return 0; }
@@ -244,8 +245,9 @@
     }
 
     // instantiate the queue...
+    WorkerPool worker_pool (4, true);
     GIOChannelSocket::Creator socket_creator;
-    Queue queue (data, data, &socket_creator, prefs.get_flag("work-online",true));
+    Queue queue (data, data, &socket_creator, worker_pool, prefs.get_flag("work-online", true));
     g_timeout_add (5000, queue_upkeep_timer_cb, &queue);
 
     if (nzb)
@@ -294,6 +296,8 @@
       run_pan_in_window (cache, data, queue, prefs, group_prefs, GTK_WINDOW(window));
     }
 
+    WorkerPool::quitAllWorkers();
+
     if (prefs.get_flag("clear-article-cache-on-shutdown", false))
       cache.clear ();
   }
Index: pan/gui/task-pane.cc
===================================================================
--- pan/gui/task-pane.cc	(revision 206)
+++ pan/gui/task-pane.cc	(working copy)
@@ -210,11 +210,11 @@
   {
     Task * task (*it);
     const Queue::TaskState state (tasks.get_state (task));
-    if (state == Queue::RUNNING)
+    if (state == Queue::RUNNING || state == Queue::DECODING)
       ++running_count;
     else if (state == Queue::STOPPED)
       ++stopped_count;
-    else if (state == Queue::QUEUED)
+    else if (state == Queue::QUEUED || state == Queue::QUEUED_FOR_DECODE)
       ++queued_count;
 
     if (state==Queue::RUNNING || state==Queue::QUEUED)
@@ -315,6 +315,8 @@
     const char * state_str (0);
     switch (state) {
       case Queue::RUNNING:  state_str = _("Running"); break;
+      case Queue::DECODING: state_str = _("Decoding"); break;
+      case Queue::QUEUED_FOR_DECODE: state_str = _("Queued for Decode"); break;
       case Queue::QUEUED:   state_str = _("Queued"); break;
       case Queue::STOPPED:  state_str = _("Stopped"); break;
       case Queue::REMOVING: state_str = _("Removing"); break;
@@ -372,7 +374,7 @@
     }
 
     char * str (0);
-    if (state == Queue::RUNNING)
+    if (state == Queue::RUNNING || state == Queue::DECODING)
       str = g_markup_printf_escaped ("<b>%s</b>\n<small>%s</small>", description.c_str(), status.c_str());
     else
       str = g_markup_printf_escaped ("%s\n<small>%s</small>", description.c_str(), status.c_str());
Index: pan/gui/gui.cc
===================================================================
--- pan/gui/gui.cc	(revision 206)
+++ pan/gui/gui.cc	(working copy)
@@ -317,6 +317,17 @@
   _prefs.add_listener (this);
 
   gtk_accel_map_load (get_accel_filename().c_str());
+
+  { // make sure taskbar views have the right tasks in them -- 
+    // when Pan first starts, the active tasks are already running
+    Queue::task_states_t task_states;
+    queue.get_all_task_states(task_states);    
+    foreach(Queue::tasks_t, task_states.tasks, it) {
+      Queue::TaskState s = task_states.get_state(*it);
+      if (s == Queue::RUNNING || s == Queue::DECODING)
+        on_queue_task_active_changed (queue, *(*it), true);
+    }
+  }
 }
 
 namespace
@@ -1142,13 +1153,13 @@
 void GUI :: do_about_pan ()
 {
 #if GTK_CHECK_VERSION(2,6,0)
-  const gchar * authors [] = { "Charles Kerr", 0 };
+  const gchar * authors [] = { "Charles Kerr <charles@rebelbase.com>", "Calin Culianu <calin@ajvar.org> - Threaded Decoding", 0 };
   GdkPixbuf * logo = gdk_pixbuf_new_from_inline(-1, icon_pan_about_logo, 0, 0);
   GtkAboutDialog * w (GTK_ABOUT_DIALOG (gtk_about_dialog_new ()));
   gtk_about_dialog_set_name (w, _("Pan"));
   gtk_about_dialog_set_version (w, PACKAGE_VERSION);
   gtk_about_dialog_set_comments (w, VERSION_TITLE);
-  gtk_about_dialog_set_copyright (w, _("Copyright © 2002-2006 Charles Kerr"));
+  gtk_about_dialog_set_copyright (w, _("Copyright © 2002-2007 Charles Kerr"));
   gtk_about_dialog_set_website (w, "http://pan.rebelbase.com/");
   gtk_about_dialog_set_logo (w, logo);
   gtk_about_dialog_set_license (w, LICENSE);
@@ -1168,7 +1179,7 @@
   gtk_box_pack_start (box, gtk_image_new_from_pixbuf (logo), false, false, PAD);
   gtk_box_pack_start (box, gtk_label_new("Pan " PACKAGE_VERSION), false, false, PAD);
   gtk_box_pack_start (box, gtk_label_new(VERSION_TITLE), false, false, 0);
-  gtk_box_pack_start (box, gtk_label_new(_("Copyright © 2002-2006 Charles Kerr")), false, false, 0);
+  gtk_box_pack_start (box, gtk_label_new(_("Copyright © 2002-2007 Charles Kerr")), false, false, 0);
   gtk_box_pack_start (box, gtk_label_new("http://pan.rebelbase.com/"), false, false, PAD);
   gtk_widget_show_all (dialog);
   g_signal_connect_swapped (dialog, "response", G_CALLBACK (gtk_widget_destroy), dialog);
Index: pan/tasks/task.cc
===================================================================
--- pan/tasks/task.cc	(revision 206)
+++ pan/tasks/task.cc	(working copy)
@@ -60,18 +60,30 @@
    }
 }
 
-#if 0
-time_t
-Task :: get_oldest_article_time () const
+/***
+****
+***/
+
+void
+Task :: give_decoder (DecoderSource* s, Decoder* d)
 {
-   time_t oldest_time (~0);
+  _decoder_to_source[d] = s;
+  use_decoder (d);
+}
+void
+Task :: use_decoder (Decoder * d)
+{
+  abort ();
+}
+void
+Task :: check_in (Decoder * d)
+{
+  decoder_to_source_t::iterator it (_decoder_to_source.find (d));
 
-   for (articles_t::const_iterator it=_articles.begin(), end=_articles.end(); it!=end; ++it) {
-      const time_t t (it->get_oldest_part_time ());
-      if (t < oldest_time)
-         oldest_time = t;
-   }
-
-   return oldest_time;
+  if (it != _decoder_to_source.end())
+  {
+    DecoderSource * s (it->second);
+    _decoder_to_source.erase (d);
+    s->check_in (d, this);
+  }
 }
-#endif
Index: pan/tasks/socket-impl-gio.cc
===================================================================
--- pan/tasks/socket-impl-gio.cc	(revision 206)
+++ pan/tasks/socket-impl-gio.cc	(working copy)
@@ -24,7 +24,6 @@
 ******/
 
 #include <config.h>
-#include <map>
 #include <iostream>
 #include <string>
 #include <cerrno>
@@ -38,6 +37,7 @@
 
 #include <pan/general/file-util.h>
 #include <pan/general/log.h>
+#include <pan/general/worker-pool.h>
 
 #ifdef G_OS_WIN32
   // this #define is necessary for mingw
@@ -91,14 +91,14 @@
 
 using namespace pan;
 
-namespace 
+namespace
 {
   typedef int (*t_getaddrinfo)(const char *,const char *, const struct addrinfo*, struct addrinfo **);
   t_getaddrinfo p_getaddrinfo (0);
 
   typedef void (*t_freeaddrinfo)(struct addrinfo*);
   t_freeaddrinfo p_freeaddrinfo (0);
-  
+
   void ensure_module_inited (void)
   {
     static bool inited (false);
@@ -107,19 +107,19 @@
     {
       p_freeaddrinfo=NULL;
       p_getaddrinfo=NULL;
-      
+
 #ifdef G_OS_WIN32
       WSADATA wsaData;
       WSAStartup(MAKEWORD(2,2), &wsaData);
-      
+
       char sysdir[MAX_PATH], path[MAX_PATH+8];
-      
+
       if(GetSystemDirectory(sysdir,MAX_PATH)!=0)
       {
         HMODULE lib=NULL;
         FARPROC pfunc=NULL;
         const char *libs[]={"ws2_32","wship6",NULL};
-        
+
         for(const char **p=libs;*p!=NULL;++p)
         {
           g_snprintf(path,MAX_PATH+8,"%s\\%s",sysdir,*p);
@@ -164,7 +164,7 @@
     signal (SIGPIPE, SIG_IGN);
 #endif
 
-    // get an addrinfo for the host 
+    // get an addrinfo for the host
     const std::string host (host_in.str, host_in.len);
     char portbuf[32];
     g_snprintf (portbuf, sizeof(portbuf), "%d", port);
@@ -178,7 +178,7 @@
 
       err = WSAGetLastError();
       if (err || !ans) {
-        setme_err = get_last_error (err);	
+        setme_err = get_last_error (err);
         return 0;
       }
 
@@ -290,7 +290,7 @@
 *****
 ****/
 
-GIOChannelSocket :: GIOChannelSocket ( ):
+GIOChannelSocket :: GIOChannelSocket ():
    _channel (0),
    _tag_watch (0),
    _tag_timeout (0),
@@ -383,7 +383,7 @@
       g_string_prepend_len (g, _partial_read.c_str(), _partial_read.size());
       _partial_read.clear ();
 
-      debug ("read [" << g->str << "]");
+      debug_v ("read [" << g->str << "]"); // verbose debug, if --debug --debug was on the command-line
       increment_xfer_byte_count (g->len);
       if (g_str_has_suffix (g->str, "\r\n"))
         g_string_truncate (g, g->len-2);
@@ -410,8 +410,8 @@
 
   return IO_DONE;
 }
- 
 
+
 GIOChannelSocket :: DoResult
 GIOChannelSocket :: do_write ()
 {
@@ -554,7 +554,8 @@
 
 namespace
 {
-  struct ThreadInfo
+  struct ThreadWorker : public WorkerPool::Worker,
+                        public WorkerPool::Worker::Listener
   {
     std::string host;
     int port;
@@ -564,44 +565,33 @@
     Socket * socket;
     std::string err;
 
-    ThreadInfo (const StringView& h, int p, Socket::Creator::Listener *l):
+    ThreadWorker (const StringView& h, int p, Socket::Creator::Listener *l):
       host(h), port(p), listener(l), ok(false), socket(0) {}
-  };
 
-  gboolean socket_created_idle (gpointer info_gpointer)
-  {
-    ThreadInfo * info (static_cast<ThreadInfo*>(info_gpointer));
-    if (!info->err.empty())
-      Log :: add_err (info->err.c_str());
-    info->listener->on_socket_created (info->host, info->port, info->ok, info->socket);
-    delete info;
-    return false;
-  }
+    void do_work (void *ignored)
+    {
+      socket = new GIOChannelSocket ();
+      ok = socket->open (host, port, err);
+    }
 
-  void create_socket_thread_func (gpointer info_gpointer, gpointer unused)
-  {
-    //std::cerr << LINE_ID << " creating a socket in worker thread...\n";
-    ThreadInfo * info (static_cast<ThreadInfo*>(info_gpointer));
-    info->socket = new GIOChannelSocket ();
-    info->ok = info->socket->open (info->host, info->port, info->err);
-    g_idle_add (socket_created_idle, info); // pass results to main thread...
-  }
+    /** called in main thread after do_work() is done */
+    void on_work_complete (void *ignored)
+    {
+      // pass results to main thread...
+      if (!err.empty())   Log :: add_err (err.c_str());
+      listener->on_socket_created (host, port, ok, socket);
+    }
+  };
 }
-  
+
 void
 GIOChannelSocket :: Creator :: create_socket (const StringView & host,
                                               int                port,
+                                              WorkerPool       & threadpool,
                                               Listener         * listener)
 {
   ensure_module_inited ();
 
-  static GThreadPool * pool (0);
-  if (!pool)
-    pool = g_thread_pool_new (create_socket_thread_func, 0, 4, true, 0);
-
-  // farm this out to a worker thread so that the main thread
-  // doesn't block while we open the socket.
-  g_thread_pool_push (pool, 
-                      new ThreadInfo (host, port, listener),
-                      NULL);
+  ThreadWorker * w = new ThreadWorker (host, port, listener);
+  threadpool.push_work (w, 0, w, true);
 }
Index: pan/tasks/socket-impl-gio.h
===================================================================
--- pan/tasks/socket-impl-gio.h	(revision 206)
+++ pan/tasks/socket-impl-gio.h	(working copy)
@@ -28,7 +28,7 @@
 {
   /**
    * glib implementation of Socket
-   * 
+   *
    * @ingroup tasks
    */
   class GIOChannelSocket: public Socket
@@ -69,7 +69,7 @@
       class Creator: public Socket::Creator {
         public:
           virtual ~Creator () { }
-          virtual void create_socket (const StringView& host, int port, Listener *l);
+          virtual void create_socket (const StringView& host, int port, WorkerPool&, Listener *l);
       };
   };
 }
Index: pan/tasks/decoder.cc
===================================================================
--- pan/tasks/decoder.cc	(revision 0)
+++ pan/tasks/decoder.cc	(revision 0)
@@ -0,0 +1,326 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2002-2007  Charles Kerr <charles@rebelbase.com>
+ *
+ * This file
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ * Copyright (C) 2007 Charles Kerr <charles@rebelbase.com>
+ *
+ * This file is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include <config.h>
+#include <algorithm>
+#include <cerrno>
+extern "C" {
+#  define PROTOTYPES
+#  include <uulib/uudeview.h>
+#  include <glib/gi18n.h>
+};
+#include <pan/general/debug.h>
+#include <pan/general/file-util.h>
+#include <pan/general/foreach.h>
+#include <pan/data/article-cache.h>
+#include "decoder.h"
+
+using namespace pan;
+
+// re-initialize the object
+void
+Decoder :: init ( TaskArticle                 * t,
+                  const Quark                 & sp,
+                  const strings_t             & f,
+                  const TaskArticle::SaveMode & sm )
+{
+  disable_progress_update();
+  task = t;
+  save_path = sp.to_string();
+  filenames = f;
+  save_mode = sm;
+  mark_read = false;
+  log_errors.clear();
+  percent = 0;
+  current_file = "";
+  log_infos.clear();
+  please_stop = false; // clear WorkerPool::worker state...
+  quit = false; // clear more WorkerPool::worker state..
+}
+
+Decoder :: Decoder (WorkerPool& pool):
+  _worker_pool (pool),
+  _gsourceid (-1)
+{
+}
+
+Decoder :: ~Decoder()
+{
+  disable_progress_update();
+}
+
+void
+Decoder :: do_work(void *ignored) // save article in another thread to avoid network stalls
+{
+  static const int bufsz = 4096;
+  char buf[bufsz];
+
+  enable_progress_update();
+
+  // NOTE THIS WHOLE METHOD RUNS IN ONE SEPARATE THREAD -- so be sure to keep that in mind..
+  if (save_mode & TaskArticle::RAW)
+  {
+    int i = 0;
+    foreach_const (ArticleCache::strings_t, filenames, it)
+    {
+      if (please_stop) break; // poll WorkerPool::Worker stop flag
+
+      gchar * contents (0);
+      gsize length (0);
+      if (g_file_get_contents (it->c_str(), &contents, &length, NULL) && length>0)
+      {
+        file :: ensure_dir_exists (save_path.c_str());
+        gchar * basename (g_path_get_basename (it->c_str()));
+        gchar * filename (g_build_filename (save_path.c_str(), basename, NULL));
+        FILE * fp = fopen (filename, "w+");
+
+        mut.lock();
+        current_file = filename; // save the filename in class so that progress code can see it potentially
+        mut.unlock();
+
+        if (!fp) {
+          g_snprintf(buf, bufsz, _("Couldn't save file \"%s\": %s"), filename, file::pan_strerror(errno));
+          log_errors.push_back (buf); // log error
+        } else {
+          fwrite (contents, 1, (size_t)length, fp);
+          fclose (fp);
+        }
+        g_free (filename);
+        g_free (basename);
+      }
+      g_free (contents);
+
+      // update percent for progress_update_timer_func
+      mut.lock();
+      percent = ++i*100/filenames.size();
+      mut.unlock();
+    }
+  }
+
+  if (save_mode & TaskArticle::DECODE)
+  {
+    // decode
+    int res, step = 0;
+    if (((res = UUInitialize())) != UURET_OK)
+      log_errors.push_back(_("Error initializing uulib")); // log error
+    else
+    {
+      UUSetMsgCallback (this, uu_log);
+      UUSetOption (UUOPT_DESPERATE, 1, NULL); // keep incompletes -- they're still useful to par2
+
+      int i (0);
+      foreach_const (ArticleCache::strings_t, filenames, it) {
+        if (please_stop) break; // poll WorkerPool::Worker stop flag
+        if ((res = UULoadFileWithPartNo (const_cast<char*>(it->c_str()), 0, 0, ++i)) != UURET_OK) {
+          g_snprintf(buf, bufsz,
+                   _("Error reading from %s: %s"),
+                   it->c_str(),
+                   (res==UURET_IOERR)
+                   ?  file::pan_strerror (UUGetOption (UUOPT_ERRNO, NULL,
+                                                       NULL, 0))
+                   : UUstrerror(res));
+          log_errors.push_back(buf); // log error
+        }
+        mut.lock();
+        // phase I of progress..
+        int tmp = percent = ++step*10/filenames.size(); // update percentage progress member .. not this goes up to 10% and the remaining 90% is calculated from the actually uuprogress given us by uulib
+        mut.unlock();
+        debug("uudecoder thread: first pass progress " << tmp << "%");
+      }
+
+
+      i = 0;
+      uulist * item;
+
+      UUSetBusyCallback (this, uu_busy_poll, 500); // .5 secs busy poll?
+
+      i = 0;
+      while ((item = UUGetFileListItem (i++)))
+      {
+        if (please_stop) break; // poll WorkerPool::Worker stop flag
+
+        // make sure the directory exists...
+        if (!save_path.empty())
+          file :: ensure_dir_exists (save_path.c_str());
+
+        // find a unique filename...
+        char * fname (0);
+        for (int i=0; ; ++i) {
+          std::string basename ((item->filename && *item->filename)
+                                ? item->filename
+                                : "pan-saved-file");
+          if (i) {
+            g_snprintf (buf, bufsz, "_copy_%d", i+1); // we don't want "_copy_1"
+            // try to preserve any extension
+            std::string::size_type dotwhere = basename.find_last_of(".");
+            if (dotwhere != basename.npos) {// if we found a dot
+          	  std::string bn (basename, 0, dotwhere); // everything before the last dot
+          	  std::string sf (basename, dotwhere, basename.npos); // the rest
+          	  // add in a substring to make it unique and enable things like "rm -f *_copy_*"
+          	  basename = bn + buf + sf;
+            }else{
+          	  basename += buf;
+            }
+          }
+          fname = save_path.empty()
+            ? g_strdup (basename.c_str())
+            : g_build_filename (save_path.c_str(), basename.c_str(), NULL);
+          if (!file::file_exists (fname))
+            break;
+          g_free (fname);
+        }
+
+        // decode the file...
+        if ((res = UUDecodeFile (item, fname)) == UURET_OK) {
+          g_snprintf(buf, bufsz,_("Saved \"%s\""), fname);
+          log_infos.push_back(buf); // log info
+        } else if (res == UURET_NODATA) {
+          // silently let this error by... user probably tried to
+          // save attachements on a text-only post
+        } else {
+          const int the_errno (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0));
+          if (res==UURET_IOERR && the_errno==ENOSPC) {
+            g_snprintf (buf, bufsz, _("Error saving \"%s\":\n%s. %s"), fname, file::pan_strerror(the_errno), "ENOSPC");
+            log_errors.push_back(buf); // log this to the error log -- used to be log_urgents but we decided against an urgent list because it can popup too many dialogs?  bug #420618 discussion..
+          } else {
+            g_snprintf (buf, bufsz,_("Error saving \"%s\":\n%s."),
+                             fname,
+                             res==UURET_IOERR ? file::pan_strerror(the_errno) : UUstrerror(res));
+            log_errors.push_back(buf); // log error
+          }
+        }
+
+        // cleanup
+        g_free (fname);
+
+      }
+
+      mark_read = true;
+    }
+    UUCleanUp ();
+  }
+
+  if (please_stop)
+    debug("got notification to stop early, decode might not be finished..");
+  disable_progress_update();
+}
+
+/* static */
+void
+Decoder :: uu_log (void* data, char* message, int severity)
+{
+    Decoder *self = reinterpret_cast<Decoder *>(data);
+    char * pch (g_locale_to_utf8 (message, -1, 0, 0, 0));
+
+    if (severity == UUMSG_PANIC || severity==UUMSG_FATAL || severity==UUMSG_ERROR)
+      self->log_errors.push_back (pch ? pch : message);
+    else if (severity == UUMSG_WARNING || severity==UUMSG_NOTE)
+      self->log_infos.push_back (pch ? pch : message);
+
+    g_free (pch);
+}
+
+/* static */
+int
+Decoder :: uu_busy_poll (void *data, uuprogress *p)
+{
+  Decoder *self = reinterpret_cast<Decoder *>(data);
+  if (self->please_stop) {
+    debug("uudecoder thread: got stop request, aborting early");
+    return 1;
+  }
+
+  debug("uudecoder thread: uuprogress is " << p->percent << " percent on " << p->fsize << "b file `" << p->curfile << "' part " << p->partno << " of " << p->numparts);
+
+  assert(p->numparts); // this should always be nonzero unless something is totally wrong
+
+  self->mut.lock();
+
+  double pct = p->percent;
+  if (p->numparts == 1) // we are in phase III of uudecode.. so pick up percent here by offsetting from 20.0% becasue Phase I and II each took 10%
+    self->percent =  int( pct = (pct/100.0) * (100.0-20.0) + 20.0 ); // complicated way to update percentage -- this is because the uudecode progress is reset to 0 for each phase so we have to fudge its value
+  else if(p->numparts) // phase II is here, it takes about 10% of total time (phase I was in the do_work function and it took also 10%)
+    self->percent = int(pct = 10 + p->partno*10/p->numparts);
+  else  { // something is wrong, this shouldn't be reached.. we have this here to guard against division by zero above
+    self->percent = int (pct); // noop
+    debug ("uudecoder thread got p->numparts == 0!  HELP!");
+  }
+
+  self->current_file = p->curfile;
+
+  self->mut.unlock();
+  
+  debug("uudecoder thread: calculated percent is " << pct);
+
+  return 0;
+}
+
+void
+Decoder :: enqueue_work_in_thread (TaskArticle                 * listener,
+                                   void                        * listener_data,
+                                   const Quark                 & save_path,
+                                   const strings_t             & filenames,
+                                   const TaskArticle::SaveMode & save_mode)
+{
+  init (listener, save_path, filenames,  save_mode);
+
+  // gentlemen, start your saving...
+  _worker_pool.push_work (this,          /* who is the worker?         */
+                          listener_data, /* void* data passed listener */
+                          listener,      /* who is the listener?       */
+                          false          /* don't auto-delete worker   */);
+}
+
+gboolean
+Decoder :: progress_update_timer_func(gpointer decoder)
+{
+  Decoder *self = reinterpret_cast<Decoder *>(decoder);
+  Task *task = self->task;
+  if (!task || self->was_cancelled()) return false;
+
+  self->mut.lock();
+  int percent = self->percent;
+  std::string f = self->current_file;
+  self->mut.unlock();
+  task->set_step(percent);
+  task->set_status_va (_("Decoding %s"), f.c_str());
+  debug("setting task progress to: " << percent << "% file: " << f);
+
+  return true; // keep timer func running
+}
+
+void
+Decoder :: enable_progress_update ()
+{
+  if (_gsourceid == -1)
+      _gsourceid = g_timeout_add(500, progress_update_timer_func, this);
+}
+
+void
+Decoder :: disable_progress_update ()
+{
+  if (_gsourceid > -1) {
+    g_source_remove (_gsourceid);
+    _gsourceid = -1;
+  }
+}
Index: pan/tasks/decoder.h
===================================================================
--- pan/tasks/decoder.h	(revision 0)
+++ pan/tasks/decoder.h	(revision 0)
@@ -0,0 +1,119 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2002-2007 Charles Kerr <charles@rebelbase.com>
+ *
+ * This file
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ * Copyright (C) 2007 Charles Kerr <charles@rebelbase.com>
+ *
+ * This file is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef _Decoder_H_
+#define _Decoder_H_
+
+#include <list>
+#include <pan/general/locking.h>
+#include <pan/general/worker-pool.h>
+#include <pan/data/article-cache.h>
+#include <pan/tasks/task-article.h>
+extern "C" {
+#  define PROTOTYPES
+#  include <uulib/uudeview.h>
+};
+
+namespace pan
+{
+  /**
+   * Decodes attachments in a worker thread.
+   *
+   * @author Calin Culianu <calin@ajvar.org>
+   * @author Charles Kerr <charles@rebelbase.com>
+   * @ingroup tasks
+   * @see Queue
+   * @see TaskArticle
+   */
+  class Decoder : protected WorkerPool::Worker
+  {
+    public:
+
+      Decoder (WorkerPool&);
+      ~Decoder ();
+
+      /**
+       * Cancel the current decode, if there is one.
+       * The listener is notified via WorkerPool::Worker::on_work_cancelled().
+       */
+      void cancel() { WorkerPool::Worker::cancel();  }
+      /**
+       * Cancel the current decode, if there is one.
+       * The listener is *NOT* notified!  
+       * Use this if quitting app or if listener is or will be deleted.
+       */
+      void gracelessly_quit() { WorkerPool::Worker::gracelessly_quit();  }
+
+      typedef ArticleCache::strings_t strings_t;
+
+      void enqueue_work_in_thread (TaskArticle                    * task,
+                                   void                           * listener_data,
+                                   const Quark                    & save_path,
+                                   const strings_t                & input_files,
+                                   const TaskArticle::SaveMode    & save_mode);
+
+    public:
+
+      TaskArticle::SaveMode save_mode;
+      std::list<std::string> log_errors, log_infos;
+      bool mark_read;
+
+    private:
+
+      std::string save_path;
+      strings_t filenames;
+
+      /* The below values are automagically polled from the main thread
+         while the decode task is running, and updates are sent to the Task. */
+      Mutex mut;
+      volatile int percent;
+      std::string current_file; // the current file we are decoding, with path
+
+      TaskArticle * task;
+
+    protected:
+
+      /// from WorkerPool::Worker interface, saves article to filesystem
+      void do_work(void *);
+
+    private:
+
+      /// just clears member vars for another run in a different thread
+      void init (TaskArticle                 * task,
+                 const Quark                 & save_path,
+                 const strings_t             & filenames,
+                 const TaskArticle::SaveMode & save_mode);
+
+      static void uu_log(void *thiz, char *message, int severity);
+      static int uu_busy_poll(void * self, uuprogress *p);
+      /// updates Progress * object (aka task) about progress of decode step
+      static gboolean progress_update_timer_func(gpointer decoder);
+
+      WorkerPool& _worker_pool;
+      int _gsourceid;
+      void disable_progress_update();
+      void enable_progress_update();
+  };
+}
+
+#endif
Index: pan/tasks/queue.cc
===================================================================
--- pan/tasks/queue.cc	(revision 206)
+++ pan/tasks/queue.cc	(working copy)
@@ -33,11 +33,15 @@
 
 Queue :: Queue (ServerInfo         & server_info,
                 TaskArchive        & archive,
-                Socket::Creator    * socket_creator, 
+                Socket::Creator    * socket_creator,
+                WorkerPool         & pool,
                 bool                 online):
   _server_info (server_info),
   _is_online (online),
   _socket_creator (socket_creator),
+  _worker_pool (pool),
+  _decoder (pool),
+  _decoder_task (0),
   _needs_saving (false),
   _last_time_saved (0),
   _archive (archive)
@@ -135,7 +139,7 @@
 
   // upkeep on running tasks... this lets us pop open
   // extra connections if the task can handle >1 connection
-  std::set<Task*> active; 
+  std::set<Task*> active;
   foreach (nntp_to_task_t, _nntp_to_task, it)
     active.insert (it->second);
   foreach (std::set<Task*>, active, it)
@@ -183,6 +187,8 @@
   std::set<Task*> active_tasks;
   foreach_const (nntp_to_task_t, _nntp_to_task, it)
     active_tasks.insert (it->second);
+  if (_decoder_task)
+    active_tasks.insert (_decoder_task);
   active = active_tasks.size ();
   total = _tasks.size ();
 }
@@ -208,6 +214,7 @@
   debug ("in process_task with a task of type " << task->get_type());
 
   const Task::State& state (task->get_state());
+  const bool was_active (task_is_active (task));
 
   if (state._work == Task::COMPLETED)
   {
@@ -222,7 +229,7 @@
   else if (_stopped.count(task))
   {
     debug ("stopped");
-    // do nothing
+    task->stop();
   }
   else if (state._health == COMMAND_FAILED)
   {
@@ -234,12 +241,20 @@
     debug ("working");
     // do nothing
   }
+  else if (state._work == Task::NEED_DECODER)
+  {
+    if (!_decoder_task)
+    {
+      _decoder_task = task;
+      task->give_decoder (this, &_decoder);
+    }
+  }
   else while (state._work == Task::NEED_NNTP)
   {
     // make the requests...
     const Task::State::unique_servers_t& servers (state._servers);
     foreach_const (Task::State::unique_servers_t, servers, it)
-      get_pool(*it).request_nntp ();
+      get_pool(*it).request_nntp (_worker_pool);
 
     Quark server;
     if (!find_best_server (servers, server))
@@ -251,6 +266,10 @@
 
     give_task_a_connection (task, nntp);
   }
+
+  const bool is_active (task_is_active (task));
+  if (is_active != was_active)
+    fire_task_active_changed (task, is_active);
 }
 
 /***
@@ -258,25 +277,33 @@
 ***/
 
 Task*
+Queue :: find_first_task_needing_decoder ()
+{
+  foreach (TaskSet, _tasks, it) {
+    const Task::State& state ((*it)->get_state ());
+    if  ((state._work == Task::NEED_DECODER)
+      && (!_stopped.count (*it))
+      && (!_removing.count (*it)))
+      return *it;
+  }
+
+  return 0;
+}
+
+Task*
 Queue :: find_first_task_needing_server (const Quark& server)
 {
-  Task * retval (0);
-
-  foreach (TaskSet, _tasks, it)
-  {
+  foreach (TaskSet, _tasks, it) {
     const Task::State& state ((*it)->get_state ());
     if  ((state._health != COMMAND_FAILED)
       && (state._work == Task::NEED_NNTP)
       && (state._servers.count(server))
       && (!_stopped.count (*it))
       && (!_removing.count (*it)))
-    {
-      retval = *it;
-      break;
-    }
+      return *it;
   }
 
-  return retval;
+  return 0;
 }
 
 bool
@@ -404,8 +431,10 @@
 {
   foreach_const (tasks_t, tasks, it) {
     Task * task (*it);
-    if (_tasks.index_of (task) != -1)
+    if (_tasks.index_of (task) != -1) {
       _stopped.insert (task);
+      process_task (task);
+    }
   }
 }
 
@@ -470,6 +499,9 @@
 bool
 Queue :: task_is_active (const Task * task) const
 {
+  if (task && task==_decoder_task)
+    return true;
+
   bool task_has_nntp (false);
   foreach_const (nntp_to_task_t, _nntp_to_task, it)
     if ((task_has_nntp = task==it->second))
@@ -490,9 +522,10 @@
   const int index (_tasks.index_of (task));
   pan_return_if_fail (index != -1);
 
-  if (task_is_active (task)) // wait for the NNTPs to finish
+  if (task_is_active (task)) // wait for the Task to finish
   {
-    debug ("can't delete this task right now because it's got server connections");
+    debug ("can't delete this task right now because it's active");
+    task->stop();
     _removing.insert (task);
   }
   else // no NNTPs working, we can remove right now.
@@ -514,9 +547,20 @@
 void
 Queue :: get_all_task_states (task_states_t& setme)
 {
-  setme.tasks.clear ();
-  setme.tasks.insert (setme.tasks.begin(), _tasks.begin(), _tasks.end());
+  setme.tasks.clear();
+  setme.tasks.reserve(_tasks.size());
 
+  std::vector<Task *> & need_decode = setme._need_decode.get_container();
+  need_decode.clear();
+  need_decode.reserve(setme.tasks.capacity());
+
+  foreach(TaskSet, _tasks, it) {
+    setme.tasks.push_back(*it);
+    if ((*it)->get_state()._work == Task::NEED_DECODER)
+      need_decode.push_back(*it);
+  }
+  setme._need_decode.sort();
+
   setme._queued.get_container() = setme.tasks;
   setme._queued.sort ();
 
@@ -529,10 +573,12 @@
   removing.insert (removing.end(), _removing.begin(), _removing.end());
 
   std::vector<Task*>& running (setme._running.get_container());
-  std::set<Task*> tmp; 
+  std::set<Task*> tmp;
   foreach (nntp_to_task_t, _nntp_to_task, it) tmp.insert (it->second);
   running.clear ();
   running.insert (running.end(), tmp.begin(), tmp.end());
+
+  setme._decoding = _decoder_task;
 }
 
 void
@@ -561,10 +607,9 @@
     Task * task (_nntp_to_task[nntp]);
     _nntp_to_task.erase (nntp);
 
-    // take care of the task's state
-    const bool is_active (task_is_active (task));
-    if (!is_active) // if it's not active anymore...
-      fire_task_active_changed (task, is_active);
+    // notify the listeners if the task isn't active anymore...
+    if (!task_is_active (task))
+      fire_task_active_changed (task, false);
 
     // return the nntp to the pool
     const Quark& servername (nntp->_server);
@@ -577,6 +622,25 @@
 }
 
 void
+Queue :: check_in (Decoder* decoder, Task* task)
+{
+  // take care of our decoder counting...
+  _decoder_task = 0;
+
+  // notify the listeners if the task isn't active anymore...
+  if (!task_is_active (task))
+    fire_task_active_changed (task, false);
+
+  // pass our worker thread on to another task
+  Task * next = find_first_task_needing_decoder ();
+  if (next && (next!=task))
+    process_task (next);
+
+  // what to do now with this task...
+  process_task (task);
+}
+
+void
 Queue :: move_up (const tasks_t& tasks)
 {
   foreach_const (tasks_t, tasks, it) {
Index: pan/tasks/nntp-pool.h
===================================================================
--- pan/tasks/nntp-pool.h	(revision 206)
+++ pan/tasks/nntp-pool.h	(working copy)
@@ -29,10 +29,12 @@
 
 namespace pan
 {
+  struct WorkerPool;
+
   /**
    * A pool of NNTP connections to a particular server.
    *
-   * @ingroup tasks 
+   * @ingroup tasks
    */
   class NNTP_Pool:
     public NNTP::Source,
@@ -69,7 +71,7 @@
 
       void add_listener (Listener * l) { _listeners.insert (l); }
       void remove_listener (Listener * l) { _listeners.erase (l); }
-      void request_nntp ();
+      void request_nntp (WorkerPool&);
 
     private: //  NNTP::Listener
       virtual void on_nntp_done (NNTP*, Health, const StringView&);
Index: pan/tasks/task.h
===================================================================
--- pan/tasks/task.h	(revision 206)
+++ pan/tasks/task.h	(working copy)
@@ -29,6 +29,8 @@
 
 namespace pan
 {
+   class Decoder;
+
    /**
     * Base class for jobs that require NNTP interaction to be completed.
     * These wait their turn in the queue until an NNTP connection is available.
@@ -50,6 +52,8 @@
             COMPLETED,
             /** Task is waiting on an nntp connection */
             NEED_NNTP,
+            /** Task waiting for a decoder */
+            NEED_DECODER,
             /** Task is running */
             WORKING
          };
@@ -85,6 +89,9 @@
                void set_need_nntp (const Quark& server) {
                   _work=NEED_NNTP; _servers.clear(); _servers.insert(server); }
 
+               void set_need_decoder () {
+                   _work = NEED_DECODER; _servers.clear(); }
+
                void set_health (Health h) {
                   _health = h; }
 
@@ -95,9 +102,17 @@
 
       public:
 
-         /** Pass a NNTP connection to the Task so it can do some work */
+         /** Loan the task a NNTP connection */
          void give_nntp (NNTP::Source*, NNTP* nntp);
 
+         struct DecoderSource {
+           virtual ~DecoderSource() {}
+           virtual void check_in (Decoder*, Task*) = 0;
+         };
+
+         /** Loan the task a Decoder */
+         void give_decoder (DecoderSource*, Decoder*);
+
       public:
 
          Task (const Quark& type, const StringView& description);
@@ -109,6 +124,9 @@
 
          virtual unsigned long get_bytes_remaining () const = 0;
 
+         /// stop a running task
+         virtual void stop () { }
+
       protected:
 
          State _state;
@@ -119,6 +137,10 @@
 
          int get_nntp_count () const { return _nntp_to_source.size(); }
 
+         virtual void use_decoder (Decoder*);
+
+         void check_in (Decoder*);
+
       private:
 
          /** What type this task is ("XOVER", "POST", "SAVE", "BODIES", etc...) */
@@ -128,7 +150,12 @@
          typedef Loki::AssocVector<NNTP*,NNTP::Source*> nntp_to_source_t;
          /** used in check_in() to remember where the nntp is to be returned */
          nntp_to_source_t _nntp_to_source;
+
+         /** typedef for _decoder_to_source */
+         typedef Loki::AssocVector<Decoder*,DecoderSource*> decoder_to_source_t;
+         /** used in check_in() to remember where the decoder is to be returned */
+         decoder_to_source_t _decoder_to_source;
    };
-};
+}
 
 #endif
Index: pan/tasks/task-article.cc
===================================================================
--- pan/tasks/task-article.cc	(revision 206)
+++ pan/tasks/task-article.cc	(working copy)
@@ -1,8 +1,12 @@
 /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
 /*
  * Pan - A Newsreader for Gtk+
- * Copyright (C) 2002-2006  Charles Kerr <charles@rebelbase.com>
+ * Copyright (C) 2002-2007  Charles Kerr <charles@rebelbase.com>
  *
+ * This File:
+ * Copyright (C) 2007 Charles Kerr <charles@rebelbase.com>
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; version 2 of the License.
@@ -20,20 +24,16 @@
 #include <config.h>
 #include <algorithm>
 #include <cassert>
-#include <cerrno>
-#include <ostream>
-#include <sstream>
 extern "C" {
-  #define PROTOTYPES
-  #include <uulib/uudeview.h>
-  #include <glib/gi18n.h>
-};
+#include <glib/gi18n.h>
+}
 #include <pan/general/debug.h>
 #include <pan/general/file-util.h>
 #include <pan/general/foreach.h>
 #include <pan/general/log.h>
 #include <pan/usenet-utils/mime-utils.h>
 #include <pan/data/article-cache.h>
+#include "decoder.h"
 #include "task-article.h"
 
 using namespace pan;
@@ -63,7 +63,7 @@
                             const Article             & article,
                             ArticleCache              & cache,
                             ArticleRead               & read,
-                            Task::Listener            * listener,
+                            Progress::Listener        * listener,
                             SaveMode                    save_mode,
                             const Quark               & save_path):
   Task (save_path.empty() ? "BODIES" : "SAVE", get_description (article, !save_path.empty())),
@@ -73,8 +73,9 @@
   _read (read),
   _article (article),
   _time_posted (article.time_posted),
-  _finished_proc_has_run (false),
-  _save_mode (save_mode)
+  _save_mode (save_mode),
+  _decoder(0),
+  _decoder_has_run (false)
 {
   cache.reserve (article.get_part_mids());
 
@@ -125,12 +126,20 @@
     set_status (article.subject.c_str());
   else
     set_status_va (_("Saving %s"), article.subject.c_str());
- 
+
   update_work ();
 }
 
 TaskArticle :: ~TaskArticle ()
 {
+  
+  if (_decoder) {
+    // NB: _decoder is active here if quitting pan .. so be sure to 
+    // be careful and notify _decoder that we mean business and are QUIT..
+    // this prevents the _decoder from calling any more methods on us (WorkerPool::Worker::Listener *)
+    _decoder->gracelessly_quit();
+  }
+  
   _cache.release (_article.get_part_mids());
 }
 
@@ -161,15 +170,22 @@
     _state.set_need_nntp (servers);
   else if (working)
     _state.set_working ();
-  else {
-    _state.set_completed ();
+  else if (_save_mode && !_decoder && !_decoder_has_run) {
+    _state.set_need_decoder ();
+    set_step(0);
+  } else if (!_save_mode || _decoder_has_run) {
+    _state.set_completed();
     set_finished (OK);
-  }
+  } else assert(0 && "hm, missed a state.");
+}
 
-  if (_state._work == COMPLETED && !_finished_proc_has_run) {
-    _finished_proc_has_run = true;
-    on_finished ();
-  }
+unsigned long
+TaskArticle :: get_bytes_remaining () const
+{
+  unsigned long bytes (0);
+  foreach_const (needed_t, _needed, it) // parts not fetched yet...
+    bytes += (it->part.bytes - it->buf.size());
+  return bytes;
 }
 
 /***
@@ -243,14 +259,13 @@
   for (it=_needed.begin(); it!=_needed.end(); ++it)
     if (it->nntp == nntp)
       break;
-  g_assert (it != _needed.end());
+  assert (it != _needed.end());
 
   //std::cerr << LINE_ID << ' ' << it->part.message_id << " from " << nntp->_server << ": " << (health==OK ? "yes" : "no") << std::endl;
 
   if (health == OK) { // if download succeeded, save it in the cache
     const StringView view (&it->buf.front(), it->buf.size());
     _cache.add (it->part.get_message_id(_article.message_id), view);
-    ++_stats[nntp->_server];
   }
 
   if (health == COMMAND_FAILED) // if server doesn't have that article...
@@ -271,137 +286,70 @@
   check_in (nntp, health);
 }
 
-namespace
+/***
+****
+***/
+
+void
+TaskArticle :: use_decoder (Decoder* decoder)
 {
-  void uu_log (void* unused, char* message, int severity)
-  {
-    char * pch (g_locale_to_utf8 (message, -1, 0, 0, 0));
+  if (_state._work != NEED_DECODER)
+    check_in (decoder);
 
-    if (severity == UUMSG_PANIC || severity==UUMSG_FATAL || severity==UUMSG_ERROR)
-      Log :: add_err (pch ? pch : message);
-    else if (severity == UUMSG_WARNING || severity==UUMSG_NOTE)
-      Log :: add_info (pch ? pch : message);
+  _decoder = decoder;
+  init_steps(100);
+  _state.set_working();
+  const Article::mid_sequence_t mids (_article.get_part_mids());
+  const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+  _decoder->enqueue_work_in_thread (this, _decoder, _save_path, filenames, _save_mode);
+  set_status_va (_("Decoding %s"), _article.subject.c_str());
+  debug ("decoder thread was free, enqueued work");
+}
 
-    g_free (pch);
-  }
+void
+TaskArticle :: stop ()
+{
+  if (_decoder)
+      _decoder->cancel();
 }
 
+// called in the main thread by WorkerPool if the worker was cancelled.
 void
-TaskArticle :: on_finished ()
+TaskArticle :: on_work_cancelled (void *data)
 {
-  const Article::mid_sequence_t mids (_article.get_part_mids());
-  const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+  Decoder * d (_decoder);
+  _decoder = 0;
+  update_work ();
+  check_in (d);
+}
 
-  if (_save_mode & RAW)
-  {
-    foreach_const (ArticleCache::strings_t, filenames, it)
-    {
-      gchar * contents (0);
-      gsize length (0);
-      if (g_file_get_contents (it->c_str(), &contents, &length, NULL) && length>0)
-      {
-        file :: ensure_dir_exists (_save_path.c_str());
-        gchar * basename (g_path_get_basename (it->c_str()));
-        gchar * filename (g_build_filename (_save_path.c_str(), basename, NULL));
-        FILE * fp = fopen (filename, "w+");
-        if (!fp)
-          Log::add_err_va (_("Couldn't save file \"%s\": %s"), filename, file::pan_strerror(errno));
-        else {
-          fwrite (contents, 1, (size_t)length, fp);
-          fclose (fp);
-        }
-        g_free (filename);
-        g_free (basename);
-      }
-      g_free (contents);
-    }
-  }
+// called in the main thread by WorkerPool if the worker completed.
+void
+TaskArticle :: on_work_complete (void *data)
+{
+  assert(_decoder);
+  if (!_decoder) return;
 
-  if (_save_mode & DECODE)
-  {
-    // decode
-    int res;
-    if (((res = UUInitialize())) != UURET_OK)
-      Log::add_err (_("Error initializing uulib"));
-    else
-    {
-      UUSetMsgCallback (NULL, uu_log);
-      UUSetOption (UUOPT_DESPERATE, 1, NULL); // keep incompletes -- they're still useful to par2
+  // the decoder is done... catch up on all housekeeping
+  // now that we're back in the main thread.
 
-      int i (0);
-      foreach_const (ArticleCache::strings_t, filenames, it) {
-        if ((res = UULoadFileWithPartNo (const_cast<char*>(it->c_str()), 0, 0, ++i)) != UURET_OK)
-          Log::add_err_va (_("Error reading from %s: %s"), it->c_str(),
-            (res==UURET_IOERR) ?  file::pan_strerror (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0)) : UUstrerror(res));
-      }
+  foreach_const(std::list<std::string>, _decoder->log_errors, it)
+    Log :: add_err(it->c_str());
+  foreach_const(std::list<std::string>, _decoder->log_infos, it)
+    Log :: add_info(it->c_str());
 
-      i = 0;
-      uulist * item;
-      while ((item = UUGetFileListItem (i++)))
-      {
-        // make sure the directory exists...
-        if (!_save_path.empty())
-          file :: ensure_dir_exists (_save_path.c_str());
+  if (_decoder->mark_read)
+    _read.mark_read(_article);
 
-        // find a unique filename...
-        char * fname (0);
-        for (int i=0; ; ++i) {
-          std::string basename ((item->filename && *item->filename)
-                                ? item->filename
-                                : "pan-saved-file");
-          if (i) {
-            char buf[32];
-            g_snprintf (buf, sizeof(buf), "_copy_%d", i+1); // we don't want "_copy_1"
-            // try to preserve any extension
-            std::string::size_type dotwhere = basename.find_last_of(".");
-            if (dotwhere != basename.npos) {// if we found a dot
-          	  std::string bn (basename, 0, dotwhere); // everything before the last dot
-          	  std::string sf (basename, dotwhere, basename.npos); // the rest
-          	  // add in a substring to make it unique and enable things like "rm -f *_copy_*"
-          	  basename = bn + buf + sf;
-            }else{
-          	  basename += buf;
-            }
-          }
-          fname = _save_path.empty()
-            ? g_strdup (basename.c_str())
-            : g_build_filename (_save_path.c_str(), basename.c_str(), NULL);
-          if (!file::file_exists (fname))
-            break;
-          g_free (fname);
-        }
+  if (!_decoder->log_errors.empty())
+    set_error (_decoder->log_errors.front());
 
-        // decode the file...
-        if ((res = UUDecodeFile (item, fname)) == UURET_OK) {
-          Log::add_info_va (_("Saved \"%s\""), fname);
-        } else if (res == UURET_NODATA) {
-          // silently let this error by... user probably tried to
-          // save attachements on a text-only post
-        } else {
-          const int the_errno (UUGetOption (UUOPT_ERRNO, NULL, NULL, 0));
-          if (res==UURET_IOERR && the_errno==ENOSPC)
-            Log::add_urgent_va (_("Error saving \"%s\":\n%s. %s"), fname, file::pan_strerror(the_errno), "ENOSPC");
-          else
-            Log::add_err_va (_("Error saving \"%s\":\n%s."),
-                             fname,
-                             res==UURET_IOERR ? file::pan_strerror(the_errno) : UUstrerror(res));
-        }
+  _state.set_completed();
+  set_step (100);
+  _decoder_has_run = true;
 
-        // cleanup
-        g_free (fname);
-      }
-
-      _read.mark_read (_article);
-    }
-    UUCleanUp ();
-  }
+  Decoder * d (_decoder);
+  _decoder = 0;
+  update_work ();
+  check_in (d);
 }
-
-unsigned long
-TaskArticle :: get_bytes_remaining () const
-{
-  unsigned long bytes (0);
-  foreach_const (needed_t, _needed, it) // parts not fetched yet...
-    bytes += (it->part.bytes - it->buf.size());
-  return bytes;
-}
Index: pan/tasks/Makefile.am
===================================================================
--- pan/tasks/Makefile.am	(revision 206)
+++ pan/tasks/Makefile.am	(working copy)
@@ -3,6 +3,7 @@
 noinst_LIBRARIES = libtasks.a
 
 libtasks_a_SOURCES = \
+  decoder.cc \
   task.cc \
   task-article.cc \
   task-groups.cc \
@@ -19,6 +20,7 @@
 noinst_HEADERS = \
   adaptable-set.cc \
   adaptable-set.h \
+  decoder.h \
   defgroup.h \
   health.h \
   task.h \
Index: pan/tasks/task-article.h
===================================================================
--- pan/tasks/task-article.h	(revision 206)
+++ pan/tasks/task-article.h	(working copy)
@@ -3,6 +3,10 @@
  * Pan - A Newsreader for Gtk+
  * Copyright (C) 2002-2006  Charles Kerr <charles@rebelbase.com>
  *
+ * This file
+ * Copyright (C) 2007 Charles Kerr <charles@rebelbase.com>
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; version 2 of the License.
@@ -20,7 +24,7 @@
 #ifndef _TaskArticle_h_
 #define _TaskArticle_h_
 
-#include <map>
+#include <pan/general/worker-pool.h>
 #include <pan/data/article.h>
 #include <pan/data/article-cache.h>
 #include <pan/data/data.h>
@@ -30,11 +34,15 @@
 
 namespace pan
 {
+  struct Decoder;
+
   /**
    * Task for downloading, and optionally decoding, articles
    * @ingroup tasks
    */
-  class TaskArticle: public Task, private NNTP::Listener
+  class TaskArticle: public Task,
+                     private NNTP::Listener,
+                     private WorkerPool::Worker::Listener
   {
     public: // life cycle
 
@@ -45,17 +53,26 @@
                    const Article      & article,
                    ArticleCache       & cache,
                    ArticleRead        & read,
-                   Task::Listener     * l=0,
+                   Progress::Listener* l=0,
                    SaveMode             save_mode = NONE,
                    const Quark        & save_path = Quark());
       virtual ~TaskArticle ();
       time_t get_time_posted () const { return _time_posted; }
       const Quark& get_save_path () const { return _save_path; }
       const Article& get_article () const { return _article; }
-        
+
     public: // Task subclass
-      virtual unsigned long get_bytes_remaining () const;
+      unsigned long get_bytes_remaining () const;
+      void stop ();
 
+      /** only call this for tasks in the NEED_DECODE state
+       * attempts to acquire the saver thread and start saving
+       * returns false if failed or true if the save process started
+       * (intended to be used with the Queue class). If true is returned,
+       * a side-effect is that the task is now in the DECODING state.
+       */
+      virtual void use_decoder (Decoder*);
+
     private: // Task subclass
       virtual void use_nntp (NNTP * nntp);
 
@@ -63,6 +80,10 @@
       virtual void on_nntp_line  (NNTP*, const StringView&);
       virtual void on_nntp_done  (NNTP*, Health, const StringView&);
 
+    private: // WorkerPool::Listener interface
+      void on_work_complete(void *);
+      void on_work_cancelled(void *);
+
     protected:
       const Quark _save_path;
       const ServerRank& _server_rank;
@@ -71,13 +92,12 @@
       quarks_t _servers;
       const Article _article;
       const time_t _time_posted;
-      void on_finished ();
 
     private: // implementation
-      bool _finished_proc_has_run;
       const SaveMode _save_mode;
-      typedef std::map<Quark,int> stats_t;
-      stats_t _stats;
+      friend class Decoder;
+      Decoder * _decoder;
+      bool _decoder_has_run;
 
     private:
       struct Needed {
@@ -93,6 +113,7 @@
       needed_t _needed;
 
       void update_work ();
+
   };
 }
 
Index: pan/tasks/nntp-pool.cc
===================================================================
--- pan/tasks/nntp-pool.cc	(revision 206)
+++ pan/tasks/nntp-pool.cc	(working copy)
@@ -166,7 +166,7 @@
     std::string user, pass;
     _server_info.get_server_auth (_server, user, pass);
     NNTP * nntp = new NNTP (_server, user, pass, socket);
-    nntp->handshake (this); // 
+    nntp->handshake (this);
   }
 }
 
@@ -247,7 +247,7 @@
 
 
 void
-NNTP_Pool :: request_nntp ()
+NNTP_Pool :: request_nntp (WorkerPool& threadpool)
 {
   int active, idle, pending, max;
   get_counts (active, idle, pending, max);
@@ -269,7 +269,7 @@
     if (_server_info.get_server_addr (_server, address, port))
     {
       ++_pending_connections;
-      _socket_creator->create_socket (address, port, this);
+      _socket_creator->create_socket (address, port, threadpool, this);
     }
   }
 }
Index: pan/tasks/queue.h
===================================================================
--- pan/tasks/queue.h	(revision 206)
+++ pan/tasks/queue.h	(working copy)
@@ -25,6 +25,7 @@
 #include <vector>
 #include <pan/general/map-vector.h>
 #include <pan/general/quark.h>
+#include <pan/tasks/decoder.h>
 #include <pan/tasks/nntp-pool.h>
 #include <pan/tasks/socket.h>
 #include <pan/tasks/adaptable-set.h>
@@ -35,6 +36,7 @@
 {
   class NNTP;
   class ServerInfo;
+  class WorkerPool;
 
   /**
    * A Queue helper that saves tasks to disk and restores them from disk.
@@ -56,11 +58,12 @@
    */
   class Queue:
     public NNTP::Source,
+    public Task::DecoderSource,
     private NNTP_Pool::Listener,
     private AdaptableSet<Task*, TaskWeakOrdering>::Listener
   {
     public:
-      Queue (ServerInfo&, TaskArchive&, Socket::Creator*, bool online);
+      Queue (ServerInfo&, TaskArchive&, Socket::Creator*, WorkerPool&, bool online);
       virtual ~Queue ();
 
       typedef std::vector<Task*> tasks_t;
@@ -95,11 +98,12 @@
       void get_full_connection_counts (std::vector<ServerConnectionCounts>& setme) const;
 
     public:
-      enum TaskState { RUNNING, QUEUED, STOPPED, REMOVING };
+      enum TaskState { QUEUED, RUNNING, DECODING, STOPPED, REMOVING,
+                       QUEUED_FOR_DECODE };
 
       /**
        * An ordered collection of tasks and their corresponding TaskState s.
-       */ 
+       */
       struct task_states_t {
         friend class Queue;
         private:
@@ -108,12 +112,16 @@
           sorted_tasks_t _stopped;
           sorted_tasks_t _running;
           sorted_tasks_t _removing;
+          sorted_tasks_t _need_decode;
+          Task * _decoding;
         public:
           tasks_t tasks;
           TaskState get_state (Task* task) const {
+            if (_decoding && (task==_decoding)) return DECODING;
             if (_removing.count(task)) return REMOVING;
             if (_stopped.count(task)) return STOPPED;
             if (_running.count(task)) return RUNNING;
+            if (_need_decode.count(task)) return QUEUED_FOR_DECODE;
             if (_queued.count(task)) return QUEUED;
             return STOPPED;
           }
@@ -143,6 +151,9 @@
     public: // inherited from NNTP::Source
       virtual void check_in (NNTP*, Health);
 
+    public: // inherited from Task::DecoderSource
+      virtual void check_in (Decoder*, Task*);
+
     private: // inherited from NNTP_Pool::Listener
       virtual void on_pool_has_nntp_available (const Quark& server);
       virtual void on_pool_error (const Quark& server, const StringView& message);
@@ -153,6 +164,7 @@
       ServerInfo& _server_info;
       bool _is_online;
       Task* find_first_task_needing_server (const Quark& server);
+      Task* find_first_task_needing_decoder ();
       bool find_best_server (const Task::State::unique_servers_t& servers, Quark& setme);
       bool task_is_active (const Task*) const;
 
@@ -162,6 +174,9 @@
       std::set<Task*> _removing;
       std::set<Task*> _stopped;
       Socket::Creator * _socket_creator;
+      WorkerPool & _worker_pool;
+      Decoder _decoder;
+      Task * _decoder_task;
 
     protected:
       virtual void fire_tasks_added  (int index, int count);
Index: pan/tasks/socket.h
===================================================================
--- pan/tasks/socket.h	(revision 206)
+++ pan/tasks/socket.h	(working copy)
@@ -25,6 +25,7 @@
 namespace pan
 {
   class StringView;
+  class WorkerPool;
 
   /**
    * Defines primitive interactions with a remote server:
@@ -63,7 +64,7 @@
       void increment_xfer_byte_count (unsigned long byte_count);
       mutable unsigned long _bytes_since_last_check;
       mutable time_t _time_of_last_check;
-      mutable double _speed_KiBps; 
+      mutable double _speed_KiBps;
       bool _abort_flag;
 
     public:
@@ -74,7 +75,7 @@
        * This is currently implemented in glib with the GIOSocketCreator,
        * but can also be implemented for unit tests or ports to other
        * libraries.
-       * 
+       *
        * @ingroup tasks
        */
       struct Creator
@@ -85,7 +86,7 @@
         };
 
         virtual ~Creator () { }
-        virtual void create_socket (const StringView& host, int port, Listener*) = 0;
+        virtual void create_socket (const StringView& host, int port, WorkerPool&, Listener*) = 0;
       };
   };
 }
Index: pan/data-impl/add-server.cc
===================================================================
--- pan/data-impl/add-server.cc	(revision 206)
+++ pan/data-impl/add-server.cc	(working copy)
@@ -56,8 +56,9 @@
 
   // initialize the queue
   TaskArchive null_task_archive;
+  WorkerPool pool;
   GIOChannelSocket::Creator _socket_creator;
-  Queue queue (data, null_task_archive, &_socket_creator, true);
+  Queue queue (data, null_task_archive, &_socket_creator, pool, true);
   queue.add_task (new TaskGroups (data, servername));
 
   // start the event loop...
Index: pan/general/worker-pool.cc
===================================================================
--- pan/general/worker-pool.cc	(revision 0)
+++ pan/general/worker-pool.cc	(revision 0)
@@ -0,0 +1,148 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2002-2007 Charles Kerr <charles@rebelbase.com>
+ *
+ * This file
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ *
+ * This file is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include <cassert>
+#include <glib/gmain.h> // for g_idle_add
+#include <pan/general/debug.h>
+#include <pan/general/foreach.h>
+#include "worker-pool.h"
+
+namespace pan
+{
+  
+  /* static */ 
+  WorkerPool::WorkerSet & 
+  WorkerPool::all_workers() 
+  {
+    static WorkerPool::WorkerSet *all_workers_ptr = new WorkerSet;
+
+    return *all_workers_ptr;
+  }
+
+  WorkerPool::Worker::Worker() : please_stop(false), quit(false)
+  {
+    all_workers().insert(this);
+  }
+
+  WorkerPool::Worker::~Worker() 
+  {
+    all_workers().erase(this);
+  }
+
+  void
+  WorkerPool::quitAllWorkers()
+  {
+    debug("notifying all workers to gracelessly quit");
+    foreach(WorkerSet, all_workers(), it)
+      (*it)->gracelessly_quit();
+  }
+
+  struct WorkerPool::Work 
+  {
+    WorkerPool *pool;
+    Worker *worker;
+    void *data;
+    Worker::Listener *listener;
+    bool delete_worker;
+
+    Work (WorkerPool *p, Worker *w, void *d, Worker::Listener *l, bool dw):
+      pool(p), worker(w), data(d), listener(l), delete_worker(dw) {}
+  };
+
+  WorkerPool::WorkerPool (int nthr, bool exclusive)
+  {
+    assert ((!exclusive || nthr>0) && "can't have unlimited exclusive threads!");
+    if (!g_thread_supported ()) g_thread_init (0);
+    tpool = g_thread_pool_new (thr_wrapper,  this, nthr, exclusive, 0);
+  }
+  
+  WorkerPool::~WorkerPool()
+  {    
+    debug("Deleting thread pool 0x" << ((void *)this)
+           << " with max threads: " << g_thread_pool_get_max_threads(tpool)
+           << " num threads: " << g_thread_pool_get_num_threads(tpool)
+           << " unprocessed: " << g_thread_pool_unprocessed(tpool));
+
+    foreach(WorkerSet, my_workers, it)
+      (*it)->cancel();
+
+    foreach(WorkSet, my_work, it)
+      (*it)->pool = 0; // this is so that when work is done, it doesn't reference us, because we will have already been deleted by then!
+
+    g_thread_pool_free (tpool, false, true); // NB: will block waiting for ALL enqueued work to complete -- this has been verified to be true even if the work doesn't yet have a thread, etc  -Calin
+  }
+
+  /* static */ 
+  void WorkerPool::thr_wrapper(gpointer data, gpointer user_data)
+  {
+    WorkerPool *self = reinterpret_cast<WorkerPool *>(user_data);
+    Work *work = reinterpret_cast<Work *>(data);
+    self->doWork(work); // call class member
+  }
+
+  /* static */ 
+  gboolean WorkerPool::finalize(gpointer data)
+  {
+    Work *work = reinterpret_cast<Work *>(data);
+
+    if (work->worker->was_gracelessly_quit()) {
+
+      /* check for was_gracelessly_quit is here because
+         race condition is possible if checking in non-main thread */
+      debug("worker was gracelessly quit, aborting without notifying listeners!");
+
+    } else {
+
+      if (work->listener) {
+        if (work->worker->was_cancelled())
+          work->listener->on_work_cancelled(work->data);
+        else
+          work->listener->on_work_complete(work->data);
+      }
+
+    }
+
+    if (work->pool) { // work->pool is set to NULL if the WorkerPool was destroyed from underneath our feet! Aieeee! ;)  In that case there's no point in erasing ourselves from its list of workers anyway..
+      work->pool->my_workers.erase(work->worker);
+      work->pool->my_work.erase(work);
+    }
+
+    if (work->delete_worker) delete work->worker;
+    delete work;
+    return false; // tell main loop not to call us again
+  }
+
+  void WorkerPool::doWork(Work *work) 
+  {
+    /* do work here .. */
+    work->worker->do_work(work->data);
+    g_idle_add(finalize, work); /* deletes work, also may call listener funcs, etc */
+  }
+
+  void WorkerPool::push_work (Worker *worker, void * data, Worker::Listener *listener, bool delworker) 
+  {
+    Work * work = new Work (this, worker, data, listener, delworker);
+    my_workers.insert (worker);
+    my_work.insert(work);
+    g_thread_pool_push (tpool,  work, 0); // ends up invoking thr_wrapper in thread    
+  }
+}
Index: pan/general/debug.h
===================================================================
--- pan/general/debug.h	(revision 206)
+++ pan/general/debug.h	(working copy)
@@ -25,6 +25,7 @@
 namespace pan
 {
   extern bool _debug_flag;
+  extern bool _debug_verbose_flag;
 }
 
 #define LINE_ID '(' << __FILE__ << ':' << __LINE__ << ':' << __func__ << ')'
@@ -35,4 +36,10 @@
       std::cerr << LINE_ID << ' ' << A << '\n'; \
   } while (0)
 
+#define debug_v(A) \
+  do { \
+    if (_debug_verbose_flag) \
+      std::cerr << LINE_ID << ' ' << A << '\n'; \
+  } while (0)
+
 #endif
Index: pan/general/worker-pool.h
===================================================================
--- pan/general/worker-pool.h	(revision 0)
+++ pan/general/worker-pool.h	(revision 0)
@@ -0,0 +1,125 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2002-2007 Charles Kerr <charles@rebelbase.com>
+ *
+ * This file
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ *
+ * This file is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef _Worker_Pool_H_
+#define _Worker_Pool_H_
+
+#include <glib/gtypes.h>
+#include <glib/gthreadpool.h>
+#include <set>
+
+namespace pan
+{
+  /**
+   * A C++ wrapper for GThreadPool.
+   *
+   * @author Calin Culianu <calin@ajvar.org>
+   * @see Queue
+   * @ingroup general
+   */
+  class WorkerPool
+  {
+    public:
+
+      /** Creates a pool of num_threads, -1 means no limit, if exclusive then 
+          the threads are always running and not shared with other pools 
+          (as in glib documentation) */
+      WorkerPool(int num_threads = -1, bool exclusive = false); 
+
+      ~WorkerPool(); ///< deletes all the threads, may or may not stop workers
+    
+      /** Call this on app exit to gracelessly notify all workers to die 
+       * and not clean up after themselves, etc */
+      static void quitAllWorkers(); 
+
+      class Worker
+      {
+        public:
+          Worker();
+          virtual ~Worker();
+
+          /** Notify of stop request -- called by stopAllWorkers(), 
+              Re-implement if you think you can force a better stop in your 
+              do_work function, otherwise it sets the flag please_stop,
+              which a particular implementation of this class may or may
+              not listen to in order to abort work early. */
+          virtual void cancel() { please_stop = true; }
+          /** Makes the worker quit ASAP and not notify any listeners.
+              Call this if you think the listeners will be destroyed or are 
+              already destroyed or when the app is exiting. */
+          virtual void gracelessly_quit() { cancel(); quit = true; }
+
+          virtual bool was_cancelled() const { return please_stop; }
+          virtual bool was_gracelessly_quit() const { return quit; }
+
+        public:
+
+          struct Listener {
+            virtual ~Listener() {};
+            /** Called in the context of the main thread after enqueued work is done.
+                `data' points to the data passed in when the work was enqueued. */
+            virtual void on_work_complete(void * data) = 0;
+            virtual void on_work_cancelled(void *data) {}
+          };
+
+        protected:
+
+          /** Re-implement in your classes to actually do the work.  Called,
+              by WorkerPool framework in a worker thread. */
+          virtual void do_work(void * data) = 0;
+
+          /** workers implementing do_work() should check this flag and if flag 
+              set, worker should try and stop what it was doing */
+          volatile bool please_stop, quit;
+          friend class WorkerPool;
+      };
+
+      /** Calls w->do_work(data) for you in the worker thread.
+          When the work is completed, the optional listener is notified
+          *from the main thread*.  Optionally you can tell this function
+          to delete the Worker when the work is done as well.  If deleting,
+          the delete is done after the listener is notified in the main thread.*/
+      void push_work(Worker *w, void * data = 0, Worker::Listener * = 0, bool delete_worker_on_completion = false);
+
+    private:
+      struct Work; 
+
+      static void thr_wrapper(gpointer data, gpointer user_data);
+      static gboolean finalize(gpointer data); /// called in main thread
+      void doWork(Work *); /**< Called in the context of the 
+                            worker thread to do work, deletes work when done */
+
+      mutable GThreadPool *tpool;
+
+      typedef std::set<Worker *> WorkerSet;
+      typedef std::set<Work *>   WorkSet;
+      WorkerSet my_workers;
+      WorkSet   my_work;
+
+      /// guarded access to all_workers set -- may initialize heap-allocated static data
+      static WorkerSet & all_workers();  
+
+      WorkerPool& operator= (WorkerPool&); // not implemented
+  };
+}
+
+#endif
Index: pan/general/Makefile.am
===================================================================
--- pan/general/Makefile.am	(revision 206)
+++ pan/general/Makefile.am	(working copy)
@@ -11,7 +11,8 @@
  progress.cc \
  quark.cc \
  string-view.cc \
- text-match.cc
+ text-match.cc \
+ worker-pool.cc
 
 noinst_HEADERS = \
  debug.h \
@@ -20,6 +21,7 @@
  file-util.h \
  foreach.h \
  line-reader.h \
+ locking.h \
  log.h \
  map-vector.h \
  messages.h \
@@ -29,7 +31,8 @@
  string-view.h \
  test.h \
  text-match.h \
- time-elapsed.h
+ time-elapsed.h \
+ worker-pool.h 
 
 noinst_PROGRAMS = \
   progress-test \
Index: pan/general/debug.cc
===================================================================
--- pan/general/debug.cc	(revision 206)
+++ pan/general/debug.cc	(working copy)
@@ -4,4 +4,5 @@
 namespace pan
 {
   bool _debug_flag = false;
+  bool _debug_verbose_flag = false;
 }
Index: pan/general/locking.h
===================================================================
--- pan/general/locking.h	(revision 0)
+++ pan/general/locking.h	(revision 0)
@@ -0,0 +1,53 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2002-2007  Charles Kerr <charles@rebelbase.com>
+ *
+ * This file
+ * Copyright (C) 2007 Calin Culianu <calin@ajvar.org>
+ *
+ * This file is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.  *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef _Mutex_h_
+#define _Mutex_h_
+
+#include <glib/gthread.h>
+
+namespace pan
+{
+  /***
+   * A C++ wrapper for GMutex.
+   * @ingroup general
+   * @author Calin Culianu <calin@ajvar.org>
+   */
+  class Mutex
+  {
+    private:
+      GMutex * m;
+      Mutex& operator= (const Mutex&); ///< unimplemented
+
+    public:
+      Mutex() {
+        if (!g_thread_supported ()) g_thread_init (0);
+        m = g_mutex_new ();
+      }
+      ~Mutex() { g_mutex_free(m); }
+      /** Block until a lock is acquired */
+      void lock() { g_mutex_lock(m); }
+      /** Unlock the mutex - make wake another thread waiting on it */
+      void unlock() { g_mutex_unlock(m); }
+  };
+}
+
+#endif
