[PATCH] Decoding in a separate thread
"Calin A. Culianu" <[email protected]>
| Newsgroups | gmane.comp.gnome.apps.pan.devel |
|---|---|
| Message-ID | <[email protected]> |
Hi all,
This patch is against svn version 205. It adds:
- a worker thread pool class that is relatively painless to use
- for now this class is used to initiate server connections and for the
decoder thread
- some locking primitives wrapped in C++ classes (wrappers around glib
objects),
- reorganizes a bit of the code,
and, most importantly:
*** Provides a decoder thread so that UUDecode operations happen in a
separate thread. No more pan stalls while it uudecodes! Yay!
Please apply this patch as follows:
1. get the svn sources:
# svn checkout http://svn.gnome.org/svn/pan2/trunk pan2
2. apply the patch:
#cd pan2; patch -p1 < pan2_svn_205_add_decoder_thread.diff
3. Then build pan2:
# CFLAGS=-g CXXFLAGS=-g ./autogen.sh && ./configure && make
Please send me any and all core dumps (I seriously doubt it will coredump)
along with the compiled pan binary (I need it for debugging symbols).
If Charles is around: let me know if you accept the patch or if anything
you think needs to be changed or cleaned up.
Cheers and thanks!!
PS: PAN rocks!
PPS: Let me know if any windows users want me to do a windows build if
you are too lazy to compile it yourself on Windows...?
-Calin
_______________________________________________
Pan-devel mailing list
[email protected]
http://lists.nongnu.org/mailman/listinfo/pan-devel
pan2_svn_205_add_decoder_thread.diff
(text/plain, 82.2 KB)
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/data-impl/add-server.cc pan2.svn-decoder-threads-works!/pan/data-impl/add-server.cc
--- pan2.svn-orig/pan/data-impl/add-server.cc 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/data-impl/add-server.cc 2007-03-19 02:40:39.000000000 -0400
@@ -56,8 +56,7 @@
// initialize the queue
TaskArchive null_task_archive;
- GIOChannelSocket::Creator _socket_creator;
- Queue queue (data, null_task_archive, &_socket_creator, true);
+ Queue queue (data, null_task_archive, true);
queue.add_task (new TaskGroups (data, servername));
// start the event loop...
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/Makefile.am pan2.svn-decoder-threads-works!/pan/general/Makefile.am
--- pan2.svn-orig/pan/general/Makefile.am 2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/Makefile.am 2007-03-18 17:52:50.000000000 -0400
@@ -8,6 +8,7 @@
file-util.cc \
log.cc \
line-reader.cc \
+ locking.cc \
progress.cc \
quark.cc \
string-view.cc \
@@ -20,6 +21,7 @@
file-util.h \
foreach.h \
line-reader.h \
+ locking.h \
log.h \
map-vector.h \
messages.h \
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/debug.cc pan2.svn-decoder-threads-works!/pan/general/debug.cc
--- pan2.svn-orig/pan/general/debug.cc 2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/debug.cc 2007-03-20 02:27:11.000000000 -0400
@@ -4,4 +4,5 @@
namespace pan
{
bool _debug_flag = false;
+ bool _debug_verbose_flag = false;
}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/debug.h pan2.svn-decoder-threads-works!/pan/general/debug.h
--- pan2.svn-orig/pan/general/debug.h 2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/debug.h 2007-03-20 02:31:05.000000000 -0400
@@ -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
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/locking.cc pan2.svn-decoder-threads-works!/pan/general/locking.cc
--- pan2.svn-orig/pan/general/locking.cc 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/general/locking.cc 2007-03-18 19:48:21.000000000 -0400
@@ -0,0 +1,128 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 "locking.h"
+#include <glib.h>
+#include <cassert>
+
+namespace pan
+{
+ struct Mutex::Impl { GMutex *mut; };
+
+ Lockable::Lockable() { if (!g_thread_supported ()) g_thread_init (NULL); }
+
+ Mutex::Mutex() { p = new Impl; p->mut = g_mutex_new(); islocked = false;}
+ Mutex::~Mutex() { g_mutex_free(p->mut); delete p; p = 0; }
+
+ void Mutex::lock() { g_mutex_lock(p->mut); islocked = true; }
+ void Mutex::unlock() { islocked = false; g_mutex_unlock(p->mut); }
+ bool Mutex::tryLock() {
+ bool ret = g_mutex_trylock(p->mut);
+ if (ret) islocked = true;
+ return ret;
+ }
+
+ struct RWLock::Impl { GStaticRWLock rw; };
+
+ RWLock::RWLock()
+ : p(new Impl), isreadlocked(false), iswritelocked(false)
+ {
+ g_static_rw_lock_init(&p->rw);
+ }
+ RWLock::~RWLock() { g_static_rw_lock_free(&p->rw); delete p; p = 0; }
+
+ void RWLock::lock()
+ {
+ g_static_rw_lock_reader_lock(&p->rw);
+ assert(!iswritelocked);
+ isreadlocked = true;
+ }
+
+ void RWLock::unlock()
+ {
+ assert(isreadlocked);
+ assert(!iswritelocked);
+ isreadlocked = false;
+ g_static_rw_lock_reader_unlock(&p->rw);
+ }
+
+ bool RWLock::tryLock()
+ {
+ bool ret = g_static_rw_lock_reader_trylock(&p->rw);
+ if (ret) { isreadlocked = true; assert(!iswritelocked); }
+ return ret;
+ }
+
+ void RWLock::writeLock()
+ {
+ g_static_rw_lock_writer_lock(&p->rw);
+ iswritelocked = true;
+ assert(!isreadlocked);
+ }
+
+ void RWLock::writeUnlock()
+ {
+ assert(iswritelocked);
+ assert(!isreadlocked);
+ iswritelocked = false;
+ g_static_rw_lock_reader_unlock(&p->rw);
+ }
+
+ bool RWLock::tryWriteLock()
+ {
+ bool ret = g_static_rw_lock_writer_trylock(&p->rw);
+ if (ret) { iswritelocked = true; assert(!isreadlocked); }
+ return ret;
+ }
+
+ struct Cond::Impl { GCond *c; };
+
+ Cond :: Cond() { p = new Impl; p->c = g_cond_new(); }
+ Cond :: ~Cond() { g_cond_free(p->c); delete p; p = 0; }
+
+ // signal sleepers
+
+ void
+ Cond :: signal() { g_cond_signal(p->c); }
+ void
+ Cond :: broadcast() { g_cond_broadcast(p->c); }
+
+ // be a sleeper
+
+ void Cond :: wait(Mutex &mut)
+ {
+ assert(mut.isLocked());
+ g_cond_wait(p->c, mut.p->mut);
+ }
+
+ /** Like wait() but optionally can time out and return false
+ if the condition was not signalled. Returns true otherwise. In
+ either case the mutex should have been acquired before entering this
+ function and will be locked upon return. */
+ bool Cond :: timedWait(Mutex &mut, unsigned time_msecs)
+ {
+ assert(mut.isLocked());
+ GTimeVal tv;
+ g_get_current_time(&tv);
+ g_time_val_add(&tv, ((glong)time_msecs)*1000L);
+ return g_cond_timed_wait(p->c, mut.p->mut, &tv);
+ }
+
+
+}
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/locking.h pan2.svn-decoder-threads-works!/pan/general/locking.h
--- pan2.svn-orig/pan/general/locking.h 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/general/locking.h 2007-03-18 19:49:17.000000000 -0400
@@ -0,0 +1,136 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 pan_colon_colon_locking_H
+#define pan_colon_colon_locking_H
+
+namespace pan
+{
+
+ ///< pure virtual lockable interface for some generic algorithms...?
+ class Lockable
+ {
+ public:
+ Lockable(); ///< c'tor here needed because it may do some initialization application-wide
+ virtual ~Lockable() {}
+ virtual void lock() = 0;
+ virtual void unlock() = 0;
+ virtual bool isLocked() const = 0;
+ virtual bool tryLock() = 0;
+ };
+
+ ///< A wrapper around GMutex that is more C++-ey
+ class Mutex : public Lockable
+ {
+ public:
+ Mutex();
+ ~Mutex();
+ void lock(); ///< lock the mutex -- blocks until it is acquired
+ void unlock(); ///< unlock the mutex -- may wake 1 other thread waiting on it
+ bool isLocked() const { return islocked; }
+ bool tryLock(); ///< may lock the mutex -- does not block and returns false if it could not lock the mutex, or true if it could
+ private:
+ struct Impl; ///< needed this type so we don't put GMutex here and thus have to pull in glib.h, etc
+ Impl *p;
+ volatile bool islocked;
+ friend class Cond;
+ };
+
+ /** A wrapper around GStaticRWLock. Note that lock() acquires the read lock
+ * and writeLock() acquires the write lock.
+ */
+ class RWLock : public Lockable
+ {
+ RWLock();
+ ~RWLock();
+
+ // lockable interface methods ---
+
+ void lock(); ///< acquire the readlock -- blocks until it is acquired
+ void unlock(); ///< releasthe the readlock, waking any potential sleepers
+ bool isLocked() const { return isreadlocked; } ///< true iff readlock is acquired
+ bool tryLock(); ///< try to acquire read lock -- does not block and returns false if it could not acquire, or true if it could
+
+ // methods specific to RWLock
+
+ void writeLock(); ///< acquire the write lock -- may block
+ void writeUnlock(); ///< release the write lock -- may wake sleepers
+ bool isWriteLocked() const { return iswritelocked; } ///< true iff write lock is acquired
+ bool tryWriteLock(); ///< true iff writelock was acquired, false otherwise -- does not block
+
+ private:
+ struct Impl; ///< needed this type so we don't put GStaicRWLock here and thus have to pull in glib.h, etc
+ Impl *p;
+ volatile bool isreadlocked, iswritelocked;
+ };
+
+ /** Convenience class that automatically locks and unlocks a lockable.
+ * Locking is done in the class constructor, and unlocking is done in the
+ * destructor. This is a convenience for automatic locking and unlocking
+ * within a C++ language scope.
+ */
+ class AutoLocker
+ {
+ public:
+ /// construct from a reference to a Mutex, locks mutex
+ AutoLocker(Lockable & l) : lock(l) { lock.lock(); }
+ /// construct from a pointer to Mutex, locks mutex
+ AutoLocker(Lockable *l) : lock(*l) { lock.lock(); }
+ /// unlocks mutex
+ ~AutoLocker() { lock.unlock(); }
+ private:
+ Lockable & lock; ///< our mutex or whatever that we auto-lock/unlock
+ };
+
+ typedef class AutoLocker MutexLocker; ///< sometimes it's easier for readability to call it a MutexLocker
+
+
+ /// wraps GCond, your basic condition functionality like in glib, pthreads, et al
+ class Cond
+ {
+ public:
+ Cond();
+ ~Cond();
+
+ // signal sleepers
+
+ void signal(); ///< signal condition, waking at most 1 sleeper
+ void broadcast(); ///< wake any and all sleepers
+
+ // be a sleeper
+
+ /** wait on a condition --
+ mutex is atomically released before sleeping then
+ acquired again when this returns */
+ void wait(Mutex &mut);
+ void wait(Mutex *m) { wait(*m); }
+ /** Like wait() but optionally can time out and return false
+ if the condition was not signalled. Returns true otherwise. In
+ either case the mutex should have been acquired before entering this
+ function and will be locked upon return. */
+ bool timedWait(Mutex &mut, unsigned time_msecs);
+ /// like above function but takes a pointer arg
+ bool timedWait(Mutex *m, unsigned t) { return timedWait(*m,t); }
+
+ private:
+ struct Impl;
+ Impl *p;
+ };
+}
+
+#endif
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/general/progress.h pan2.svn-decoder-threads-works!/pan/general/progress.h
--- pan2.svn-orig/pan/general/progress.h 2007-03-20 06:17:59.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/general/progress.h 2007-03-19 02:14:41.000000000 -0400
@@ -56,10 +56,6 @@
private:
- typedef std::set<Listener*> listeners_t;
- typedef Progress::listeners_t::const_iterator listeners_cit;
- listeners_t _listeners;
-
void fire_pulse ();
void fire_percentage (int p);
void fire_status (const StringView& msg);
@@ -68,6 +64,10 @@
protected:
+ typedef std::set<Listener*> listeners_t;
+ typedef Progress::listeners_t::const_iterator listeners_cit;
+ listeners_t _listeners;
+
std::string _description; // used for default describe()
std::string _status_text; // the last status text emitted
std::vector<std::string> _errors; // the emitted error strings
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/gui.cc pan2.svn-decoder-threads-works!/pan/gui/gui.cc
--- pan2.svn-orig/pan/gui/gui.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/gui.cc 2007-03-20 05:48:01.000000000 -0400
@@ -33,6 +33,7 @@
#include <pan/tasks/task-groups.h>
#include <pan/tasks/task-xover.h>
#include <pan/tasks/nzb.h>
+#include <pan/tasks/worker-pool.h>
#include <pan/icons/pan-pixbufs.h>
#include "actions.h"
#include "body-pane.h"
@@ -317,6 +318,17 @@
_prefs.add_listener (this);
gtk_accel_map_load (get_accel_filename().c_str());
+
+ { // make sure taskbar views have the right tasks in them -- this is because 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 +1154,13 @@
void GUI :: do_about_pan ()
{
#if GTK_CHECK_VERSION(2,6,0)
- const gchar * authors [] = { "Charles Kerr", 0 };
+ const gchar * authors [] = { "Charles Kerr", "Calin Culianu <[email protected]>", 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-2006 Charles Kerr\nCopyright © 2007 Calin Culianu"));
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 +1180,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-2006 Charles Kerr\nCopyright © 2007 Calin Culianu")), 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);
@@ -1415,6 +1427,7 @@
void GUI :: do_quit ()
{
+ WorkerPool::cancelAllWorkers();
gtk_main_quit ();
}
void GUI :: do_read_selected_group ()
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/pan.cc pan2.svn-decoder-threads-works!/pan/gui/pan.cc
--- pan2.svn-orig/pan/gui/pan.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/pan.cc 2007-03-20 02:28:25.000000000 -0400
@@ -60,6 +60,7 @@
void mainloop_quit ()
{
+ WorkerPool::cancelAllWorkers(); // tell any potentially running worker pool workers to stop asap
if (nongui_gmainloop)
g_main_loop_quit (nongui_gmainloop);
else
@@ -92,6 +93,7 @@
void destroy_cb (GtkWidget*w, gpointer user_data)
{
+ WorkerPool::cancelAllWorkers(); // tell any potentially running worker pool workers to stop asap
gtk_main_quit ();
}
@@ -183,7 +185,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 +201,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 +246,7 @@
}
// instantiate the queue...
- GIOChannelSocket::Creator socket_creator;
- Queue queue (data, data, &socket_creator, prefs.get_flag("work-online",true));
+ Queue queue (data, data, prefs.get_flag("work-online",true));
g_timeout_add (5000, queue_upkeep_timer_cb, &queue);
if (nzb)
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/gui/task-pane.cc pan2.svn-decoder-threads-works!/pan/gui/task-pane.cc
--- pan2.svn-orig/pan/gui/task-pane.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/gui/task-pane.cc 2007-03-20 05:07:12.000000000 -0400
@@ -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());
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/Makefile.am pan2.svn-decoder-threads-works!/pan/tasks/Makefile.am
--- pan2.svn-orig/pan/tasks/Makefile.am 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/Makefile.am 2007-03-19 15:17:14.000000000 -0400
@@ -3,8 +3,10 @@
noinst_LIBRARIES = libtasks.a
libtasks_a_SOURCES = \
+ globals.cc \
task.cc \
task-article.cc \
+ task-article-decoder.cc \
task-groups.cc \
task-post.cc \
task-xover.cc \
@@ -14,15 +16,18 @@
socket.cc \
socket-impl-gio.cc \
socket-impl-scripted.cc \
- nntp-pool.cc
+ nntp-pool.cc \
+ worker-pool.cc
noinst_HEADERS = \
adaptable-set.cc \
adaptable-set.h \
defgroup.h \
+ globals.h \
health.h \
task.h \
task-article.h \
+ task-article-decoder.h \
task-groups.h \
task-post.h \
task-weak-ordering.h \
@@ -33,7 +38,8 @@
socket.h \
socket-impl-gio.h \
socket-impl-scripted.h \
- nntp-pool.h
+ nntp-pool.h \
+ worker-pool.h
noinst_PROGRAMS = \
adaptable-set-test \
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/globals.cc pan2.svn-decoder-threads-works!/pan/tasks/globals.cc
--- pan2.svn-orig/pan/tasks/globals.cc 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/globals.cc 2007-03-18 23:46:51.000000000 -0400
@@ -0,0 +1,62 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 "globals.h"
+
+namespace {
+
+ pan::WorkerPool *worker_pool = 0;
+ pan::GIOChannelSocket::Creator *socket_creator = 0;
+ pan::Mutex *mutex = 0;
+
+ struct Deleter {
+ ~Deleter() {
+ // NB to noobies: it's ok to delete 0
+ delete worker_pool;
+ delete socket_creator;
+ delete mutex;
+ }
+ } deleter; // d'tor called on app exit, which deletes heap-allocated objects
+
+}
+
+namespace pan
+{
+ namespace globals
+ {
+ WorkerPool & workerPool()
+ {
+ if (!worker_pool)
+ // hard-coded parameters for the worker pool...
+ worker_pool = new WorkerPool(4, true);
+ return *worker_pool;
+ }
+
+ GIOChannelSocket::Creator & socketCreator()
+ {
+ if (!socket_creator) socket_creator = new GIOChannelSocket::Creator;
+ return *socket_creator;
+ }
+
+ Mutex & mutex()
+ {
+ if (!::mutex) ::mutex = new Mutex;
+ return *::mutex;
+ }
+ }
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/globals.h pan2.svn-decoder-threads-works!/pan/tasks/globals.h
--- pan2.svn-orig/pan/tasks/globals.h 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/globals.h 2007-03-18 23:46:26.000000000 -0400
@@ -0,0 +1,42 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 pan_colon_colon_globals_H
+#define pan_colon_colon_globals_H
+
+#include <pan/tasks/worker-pool.h>
+#include <pan/tasks/socket-impl-gio.h>
+#include <pan/general/locking.h>
+
+namespace pan
+{
+ /// some global instances that can be shared and used throughout the application
+ namespace globals
+ {
+ /// useful to dole out work to slave threads
+ extern WorkerPool & workerPool();
+
+ /// a pointer to this used to be passed around to objects -- better to just make it global
+ extern GIOChannelSocket::Creator & socketCreator();
+
+ /// mutex that can be used for very coarse-grained locking appliction-wide
+ extern Mutex & mutex();
+ }
+}
+#endif
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/nntp-pool.cc pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.cc
--- pan2.svn-orig/pan/tasks/nntp-pool.cc 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.cc 2007-03-18 23:48:05.000000000 -0400
@@ -24,6 +24,7 @@
#include <pan/general/foreach.h>
#include <pan/general/log.h>
#include "nntp-pool.h"
+#include "globals.h"
using namespace pan;
@@ -37,11 +38,9 @@
}
NNTP_Pool :: NNTP_Pool (const Quark & server,
- ServerInfo & server_info,
- Socket::Creator * creator):
+ ServerInfo & server_info):
_server_info (server_info),
_server (server),
- _socket_creator (creator),
_pending_connections (0),
_active_count (0),
_time_to_allow_new_connections (0)
@@ -269,7 +268,7 @@
if (_server_info.get_server_addr (_server, address, port))
{
++_pending_connections;
- _socket_creator->create_socket (address, port, this);
+ globals::socketCreator().create_socket (address, port, this);
}
}
}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/nntp-pool.h pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.h
--- pan2.svn-orig/pan/tasks/nntp-pool.h 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/nntp-pool.h 2007-03-18 18:19:20.000000000 -0400
@@ -42,8 +42,7 @@
public:
NNTP_Pool (const Quark & server,
- ServerInfo & server_info,
- Socket::Creator *);
+ ServerInfo & server_info);
virtual ~NNTP_Pool ();
@@ -90,7 +89,6 @@
ServerInfo& _server_info;
const Quark _server;
- Socket::Creator * _socket_creator;
int _pending_connections;
struct PoolItem {
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/queue.cc pan2.svn-decoder-threads-works!/pan/tasks/queue.cc
--- pan2.svn-orig/pan/tasks/queue.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/queue.cc 2007-03-20 05:13:34.000000000 -0400
@@ -33,11 +33,9 @@
Queue :: Queue (ServerInfo & server_info,
TaskArchive & archive,
- Socket::Creator * socket_creator,
bool online):
_server_info (server_info),
_is_online (online),
- _socket_creator (socket_creator),
_needs_saving (false),
_last_time_saved (0),
_archive (archive)
@@ -75,6 +73,11 @@
fire_queue_error (message);
}
+void Queue :: on_task_decode_error(Task *task, const StringView &message)
+{
+ fire_queue_error(message);
+}
+
NNTP_Pool&
Queue :: get_pool (const Quark& servername)
{
@@ -87,7 +90,7 @@
}
else // have to build one
{
- pool = new NNTP_Pool (servername, _server_info, _socket_creator);
+ pool = new NNTP_Pool (servername, _server_info);
pool->add_listener (this);
_pools[servername] = pool;
}
@@ -135,10 +138,20 @@
// 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_or_decod;
foreach (nntp_to_task_t, _nntp_to_task, it)
- active.insert (it->second);
- foreach (std::set<Task*>, active, it)
+ active_or_decod.insert (it->second);
+ // also manage decoding tasks here
+ // upkeep on decodeing tasks.. this has the potential to move tasks
+ // from the 'need_decode' state to the decoding state
+ foreach_const (TaskSet, _tasks, it) { // need to use _tasks, as tmp above might have deleted entries
+ Task * task = *it;
+ const Task::State & state (task->get_state());
+ if (state._work==Task::DECODING || state._work == Task::NEED_DECODE)
+ active_or_decod.insert(task);
+ }
+ // process active or decoding-related tasks..
+ foreach (std::set<Task*>, active_or_decod, it)
process_task (*it);
// idle socket upkeep
@@ -183,6 +196,9 @@
std::set<Task*> active_tasks;
foreach_const (nntp_to_task_t, _nntp_to_task, it)
active_tasks.insert (it->second);
+ foreach_const (TaskSet, _tasks, it)
+ if ((*it)->get_state()._work == Task::DECODING) // decoding is considered to be active?
+ active_tasks.insert(*it);
active = active_tasks.size ();
total = _tasks.size ();
}
@@ -191,7 +207,7 @@
void
Queue :: give_task_a_connection (Task * task, NNTP * nntp)
{
- const bool was_active (task_is_active (task));
+ const bool was_active (task_was_active (task));
_nntp_to_task[nntp] = task; // it's active now...
if (!was_active)
fire_task_active_changed (task, true);
@@ -202,7 +218,7 @@
void
Queue :: process_task (Task * task)
-{
+{
pan_return_if_fail (task!=0);
debug ("in process_task with a task of type " << task->get_type());
@@ -222,7 +238,9 @@
else if (_stopped.count(task))
{
debug ("stopped");
- // do nothing
+ if (state._work == Task::DECODING)
+ // notify decoder thread of stop request
+ task->decode_cancel();
}
else if (state._health == COMMAND_FAILED)
{
@@ -234,6 +252,16 @@
debug ("working");
// do nothing
}
+ else if (state._work == Task::DECODING) {
+ debug ("decoding");
+ // do nothing
+ }
+ else if (state._work == Task::NEED_DECODE) {
+ debug ("need decode");
+ // try to save..
+ task->try_decode();
+
+ }
else while (state._work == Task::NEED_NNTP)
{
// make the requests...
@@ -251,6 +279,9 @@
give_task_a_connection (task, nntp);
}
+ const bool is_active = task_is_active(task), was_active = task_was_active(task);
+ if (is_active != was_active) // notify listeners that active may have changed
+ fire_task_active_changed (task, is_active);
}
/***
@@ -384,6 +415,7 @@
void
Queue :: fire_task_active_changed (Task * task, bool active)
{
+ _activemap[task] = active;
for (lit it(_listeners.begin()), end(_listeners.end()); it!=end; )
(*it++)->on_queue_task_active_changed (*this, *task, active);
}
@@ -468,13 +500,26 @@
}
bool
+Queue :: task_was_active(const Task *task) const
+{
+ activemap_t::const_iterator it = _activemap.find(task);
+ return it == _activemap.end() ? false : it->second;
+}
+
+bool
Queue :: task_is_active (const Task * task) const
{
bool task_has_nntp (false);
foreach_const (nntp_to_task_t, _nntp_to_task, it)
if ((task_has_nntp = task==it->second))
break;
- return task_has_nntp;
+ return task_has_nntp || task_is_decoding(task);
+}
+
+bool
+Queue :: task_is_decoding (const Task * task) const
+{
+ return task->get_state()._work == Task::DECODING;
}
void
@@ -490,9 +535,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 NNTPs or decode threads 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 got server connections or is decoding");
+ if (task_is_decoding(task)) task->decode_cancel();
_removing.insert (task);
}
else // no NNTPs working, we can remove right now.
@@ -504,6 +550,8 @@
_stopped.erase (task);
_removing.erase (task);
_tasks.remove (index);
+ if (_activemap[task]) fire_task_active_changed(task, false); // this is needed because sometimes the task is deleted and process_task may erroneously fire task_active_changed after this point...?
+ _activemap.erase(task);
delete task;
}
@@ -533,6 +581,20 @@
foreach (nntp_to_task_t, _nntp_to_task, it) tmp.insert (it->second);
running.clear ();
running.insert (running.end(), tmp.begin(), tmp.end());
+
+ tmp.clear();
+ std::vector<Task *> & decoding (setme._decoding.get_container());
+ decoding.clear();
+ foreach (tasks_t, _tasks, it) if ((*it)->get_state()._work == Task::DECODING)
+ tmp.insert (*it);
+ decoding.insert(decoding.begin(), tmp.begin(), tmp.end());
+
+ tmp.clear();
+ std::vector<Task *> & decoding_q (setme._decoding_q.get_container());
+ decoding_q.clear();
+ foreach (tasks_t, _tasks, it) if ((*it)->get_state()._work == Task::NEED_DECODE)
+ tmp.insert (*it);
+ decoding_q.insert(decoding_q.begin(), tmp.begin(), tmp.end());
}
void
@@ -562,8 +624,8 @@
_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...
+ const bool is_active (task_is_active (task)), was_active(task_was_active(task));
+ if (is_active != was_active) // if it's not active anymore...
fire_task_active_changed (task, is_active);
// return the nntp to the pool
@@ -676,3 +738,9 @@
setme_KiBps = KiBps;
setme_connections = connections;
}
+
+void
+Queue :: on_task_finished_decoding(Task *task)
+{
+ upkeep(); // call upkeep here to potentially give other tasks in the NEED_DECODE state the opportunity to start a decode going
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/queue.h pan2.svn-decoder-threads-works!/pan/tasks/queue.h
--- pan2.svn-orig/pan/tasks/queue.h 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/queue.h 2007-03-20 05:20:52.000000000 -0400
@@ -57,10 +57,11 @@
class Queue:
public NNTP::Source,
private NNTP_Pool::Listener,
- private AdaptableSet<Task*, TaskWeakOrdering>::Listener
+ private AdaptableSet<Task*, TaskWeakOrdering>::Listener,
+ private Task::Listener
{
public:
- Queue (ServerInfo&, TaskArchive&, Socket::Creator*, bool online);
+ Queue (ServerInfo&, TaskArchive&, bool online);
virtual ~Queue ();
typedef std::vector<Task*> tasks_t;
@@ -95,7 +96,7 @@
void get_full_connection_counts (std::vector<ServerConnectionCounts>& setme) const;
public:
- enum TaskState { RUNNING, QUEUED, STOPPED, REMOVING };
+ enum TaskState { RUNNING, QUEUED, STOPPED, REMOVING, DECODING, QUEUED_FOR_DECODE };
/**
* An ordered collection of tasks and their corresponding TaskState s.
@@ -108,12 +109,16 @@
sorted_tasks_t _stopped;
sorted_tasks_t _running;
sorted_tasks_t _removing;
+ sorted_tasks_t _decoding;
+ sorted_tasks_t _decoding_q;
public:
tasks_t tasks;
TaskState get_state (Task* task) const {
if (_removing.count(task)) return REMOVING;
if (_stopped.count(task)) return STOPPED;
if (_running.count(task)) return RUNNING;
+ if (_decoding.count(task)) return DECODING;
+ if (_decoding_q.count(task)) return QUEUED_FOR_DECODE;
if (_queued.count(task)) return QUEUED;
return STOPPED;
}
@@ -147,6 +152,10 @@
virtual void on_pool_has_nntp_available (const Quark& server);
virtual void on_pool_error (const Quark& server, const StringView& message);
+ private: // inherited from Task::Listener
+ void on_task_finished_decoding(Task *);
+ void on_task_decode_error(Task *task, const StringView &msg);
+
protected:
void process_task (Task *);
void give_task_a_connection (Task*, NNTP*);
@@ -155,13 +164,16 @@
Task* find_first_task_needing_server (const Quark& server);
bool find_best_server (const Task::State::unique_servers_t& servers, Quark& setme);
bool task_is_active (const Task*) const;
+ bool task_is_decoding (const Task*) const;
+ bool task_was_active (const Task*) const;
typedef std::map<NNTP*,Task*> nntp_to_task_t;
nntp_to_task_t _nntp_to_task;
std::set<Task*> _removing;
std::set<Task*> _stopped;
- Socket::Creator * _socket_creator;
+ typedef std::map<const Task *, bool> activemap_t;
+ activemap_t _activemap;
protected:
virtual void fire_tasks_added (int index, int count);
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/socket-impl-gio.cc pan2.svn-decoder-threads-works!/pan/tasks/socket-impl-gio.cc
--- pan2.svn-orig/pan/tasks/socket-impl-gio.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/socket-impl-gio.cc 2007-03-20 02:29:46.000000000 -0400
@@ -38,6 +38,7 @@
#include <pan/general/file-util.h>
#include <pan/general/log.h>
+#include "globals.h"
#ifdef G_OS_WIN32
// this #define is necessary for mingw
@@ -383,7 +384,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);
@@ -554,7 +555,8 @@
namespace
{
- struct ThreadInfo
+ struct ThreadWorker : public WorkerPool::Worker,
+ public WorkerPool::Listener
{
std::string host;
int port;
@@ -564,27 +566,29 @@
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) {}
+
+ void do_work(void *ignored); // called in thread, initiate connections in thread to avoid blocking pan when connections fail or are slow
+ void on_work_complete(void *ignored); // called in main thread when done, passes results to rest of app
};
- gboolean socket_created_idle (gpointer info_gpointer)
+
+ void ThreadWorker::on_work_complete (void *ignored)
{
- 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;
+ // pass results to main thread...
+ (void) ignored;
+ if (!err.empty()) Log :: add_err (err.c_str());
+ listener->on_socket_created (host, port, ok, socket);
+ // NB: WorkerPool framework auto-deletes us after returning from this function
}
- void create_socket_thread_func (gpointer info_gpointer, gpointer unused)
+ void ThreadWorker::do_work (void *ignored)
{
+ (void)ignored;
//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...
+ socket = new GIOChannelSocket ();
+ ok = socket->open (host, port, err);
}
}
@@ -595,13 +599,6 @@
{
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);
+ globals::workerPool().push_work(w, 0, w, true);
}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article-decoder.cc pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.cc
--- pan2.svn-orig/pan/tasks/task-article-decoder.cc 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.cc 2007-03-20 04:08:34.000000000 -0400
@@ -0,0 +1,343 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 <cassert>
+#include <cerrno>
+#include <ostream>
+#include <sstream>
+#include <list>
+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/general/locking.h>
+#include <pan/general/log.h>
+#include <pan/usenet-utils/mime-utils.h>
+#include <pan/data/article-cache.h>
+#include "task-article-decoder.h"
+
+using namespace pan;
+
+//static
+TaskArticle::Decoder TaskArticle::Decoder::instance;
+bool TaskArticle::Decoder::checked_out = false;
+
+TaskArticle::Decoder *
+TaskArticle::Decoder::check_out()
+{
+ if (!checked_out) {
+ checked_out = true;
+ return &instance;
+ }
+ return 0;
+}
+
+void
+TaskArticle::Decoder::check_in(Decoder *d)
+{
+ (void)d;
+ assert (d == &instance);
+ assert (checked_out);
+ checked_out = false;
+}
+
+// re-initialize the object which gets re-used a lot
+void
+TaskArticle :: Decoder :: init ( TaskArticle *t,
+ const Quark & sp,
+ const ArticleCache::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_urgents.clear();
+ log_errors.clear();
+ log_infos.clear();
+ base_percent = percent = 0;
+ please_stop = false; // clear WorkerPool::worker state...
+}
+
+TaskArticle :: Decoder :: Decoder()
+{
+ gsourceid = -1;
+}
+
+TaskArticle :: Decoder :: ~Decoder()
+{
+ disable_progress_update();
+}
+
+void
+TaskArticle :: Decoder :: do_work(void *ignored) // save article in another thread to avoid network stalls
+{
+ static const int bufsz = 4096;
+ char buf[bufsz];
+
+ (void)ignored;
+ 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();
+ int tmp = percent = base_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, 1000); // 1.0 secs busy poll?
+
+ i = 0;
+ while ((item = UUGetFileListItem (i++)))
+ {
+ if (please_stop) break; // poll WorkerPool::Worker stop flag
+
+ mut.lock();
+ ++itemno; // save it in class
+ mut.unlock();
+
+ // 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_urgents.push_back(buf); // log this to the urgent log
+ } 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
+TaskArticle :: Decoder :: uu_log (void* thiz, char* message, int severity)
+{
+ Decoder *self = reinterpret_cast<Decoder *>(thiz);
+ 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
+TaskArticle :: Decoder :: uu_busy_poll(void *data, uuprogress *p)
+{
+ Decoder *thiz = reinterpret_cast<Decoder *>(data);
+ if (thiz->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);
+ thiz->mut.lock();
+ double pct = p->percent, nitems = p->numparts, basepct = thiz->base_percent, item = p->partno;
+ if (p->numparts == 1) // we are in phase II of uudecode.. so pick up percent here by offsetting from remembered base_percent value computed in previous phases
+ thiz->percent = int( pct = (pct/100.0) * (100.0-basepct) + basepct ); // 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 { // phase 1 takes about 10% of total time
+ thiz->percent = thiz->base_percent = int(pct = 10.0 + p->partno*10/p->numparts);
+ }
+ thiz->current_file = p->curfile;
+ thiz->mut.unlock();
+ debug("uudecoder thread: calculated percent is " << pct);
+
+ return 0;
+}
+
+void
+TaskArticle::Decoder::enqueue_work_in_thread(TaskArticle *listener,
+ void *listener_data,
+ const Quark & save_path,
+ const ArticleCache::strings_t & filenames,
+ const TaskArticle::SaveMode & save_mode)
+{
+ init(listener, save_path, filenames, save_mode);
+
+ // gentlemen, start your saving...
+ globals::workerPool().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
+TaskArticle::Decoder::progress_update_timer_func(gpointer decoder)
+{
+ Decoder *thiz = reinterpret_cast<Decoder *>(decoder);
+ Task *task = thiz->task;
+ if (!task || thiz->was_cancelled()) return false;
+
+ thiz->mut.lock();
+ int percent = thiz->percent;
+ std::string f = thiz->current_file;
+ thiz->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 TaskArticle::Decoder::enable_progress_update()
+{
+ disable_progress_update(); // disable any running gsources, if any
+ gsourceid = g_timeout_add(1000, progress_update_timer_func, this);
+}
+
+void TaskArticle::Decoder::disable_progress_update()
+{
+ if (gsourceid > -1) {
+ g_source_remove (gsourceid);
+ gsourceid = -1;
+ }
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article-decoder.h pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.h
--- pan2.svn-orig/pan/tasks/task-article-decoder.h 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article-decoder.h 2007-03-20 01:13:49.000000000 -0400
@@ -0,0 +1,107 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 task_article_decoder_H
+#define task_article_decoder_H
+
+#include <pan/general/debug.h>
+#include <pan/general/file-util.h>
+#include <pan/general/foreach.h>
+#include <pan/general/locking.h>
+#include <pan/general/log.h>
+#include <pan/usenet-utils/mime-utils.h>
+#include <pan/data/article-cache.h>
+extern "C" {
+# define PROTOTYPES
+# include <uulib/uudeview.h>
+# include <glib/gi18n.h>
+};
+#include <glib.h>
+#include "task-article.h"
+#include "globals.h"
+
+namespace pan {
+
+ class TaskArticle::Decoder : protected WorkerPool::Worker
+ {
+ public:
+
+ void enqueue_work_in_thread(TaskArticle *task,
+ void *listener_data,
+ const Quark & save_path,
+ const ArticleCache::strings_t & filenames,
+ const TaskArticle::SaveMode & save_mode);
+
+ ~Decoder();
+
+
+ std::string save_path;
+ ArticleCache::strings_t filenames;
+ TaskArticle::SaveMode save_mode;
+
+ std::list<std::string> log_urgents, log_errors, log_infos;
+ bool mark_read;
+
+ /* The below values are automagically polled from the main thread
+ while the decode task is running, and updates are sent to the Task *
+ object this decoder owns. */
+ Mutex mut;
+ volatile int base_percent, percent, itemno, num_items;
+ std::string current_file; // the current file we are decoding, with path
+
+ TaskArticle *task;
+
+ /// may return null if the singleton is already checked out
+ static Decoder *check_out();
+
+ /** check back in the singleton that was previously checked out --
+ so that future check_outs will succeed */
+ static void check_in(Decoder *);
+
+ /// cancel an already-running decoder -- note the listener is notified with a on_work_cancelled call rather than a on_work_complete call -- from WorkerPool::Worker interface
+ void cancel() { WorkerPool::Worker::cancel(); }
+
+ protected:
+ /// from WorkerPool::Worker interface, saves article to filesystem
+ void do_work(void *);
+ /// just clears member vars for another run in a different thread
+ void init(TaskArticle *t,
+ const Quark & save_path,
+ const ArticleCache::strings_t & filenames,
+ const TaskArticle::SaveMode & save_mode);
+
+ private:
+ static void uu_log(void *thiz, char *message, int severity);
+ static int uu_busy_poll(void *thiz, uuprogress *p);
+ /// updates Progress * object (aka task) about progress of decode step
+ static gboolean progress_update_timer_func(gpointer decoder);
+ static bool checked_out; // only 1 thing can check this out at once
+
+ // singleton
+ static Decoder instance;
+
+ Decoder(); // singleton
+
+ int gsourceid;
+ void disable_progress_update();
+ void enable_progress_update();
+ };
+
+}
+
+#endif
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article.cc pan2.svn-decoder-threads-works!/pan/tasks/task-article.cc
--- pan2.svn-orig/pan/tasks/task-article.cc 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article.cc 2007-03-20 05:15:04.000000000 -0400
@@ -23,10 +23,11 @@
#include <cerrno>
#include <ostream>
#include <sstream>
+#include <list>
extern "C" {
- #define PROTOTYPES
- #include <uulib/uudeview.h>
- #include <glib/gi18n.h>
+# define PROTOTYPES
+# include <uulib/uudeview.h>
+# include <glib/gi18n.h>
};
#include <pan/general/debug.h>
#include <pan/general/file-util.h>
@@ -34,7 +35,9 @@
#include <pan/general/log.h>
#include <pan/usenet-utils/mime-utils.h>
#include <pan/data/article-cache.h>
+#include "globals.h"
#include "task-article.h"
+#include "task-article-decoder.h"
using namespace pan;
@@ -63,7 +66,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())),
@@ -74,7 +77,8 @@
_article (article),
_time_posted (article.time_posted),
_finished_proc_has_run (false),
- _save_mode (save_mode)
+ _save_mode (save_mode),
+ _decoder(0)
{
cache.reserve (article.get_part_mids());
@@ -126,13 +130,14 @@
else
set_status_va (_("Saving %s"), article.subject.c_str());
+ if (_needed.empty() && all_bytes) {
+ _state.set_need_decode();
+ set_step(0);
+ }
+
update_work ();
}
-TaskArticle :: ~TaskArticle ()
-{
- _cache.release (_article.get_part_mids());
-}
void
TaskArticle :: update_work ()
@@ -161,10 +166,8 @@
_state.set_need_nntp (servers);
else if (working)
_state.set_working ();
- else {
- _state.set_completed ();
+ else if (_state._work == COMPLETED) // this was auto-set by saver process
set_finished (OK);
- }
if (_state._work == COMPLETED && !_finished_proc_has_run) {
_finished_proc_has_run = true;
@@ -259,9 +262,15 @@
// if we got it or still have other options, we're okay
_state.set_health (health==COMMAND_FAILED && it->xref.empty() ? COMMAND_FAILED : OK);
- if (health==OK)
+ if (health==OK) {
_needed.erase (it);
- else {
+ if (_needed.empty() && _save_mode) { // if we just successfully finished with all needed parts and we need to decode the attachment, set status here
+ _state.set_need_decode();
+ set_step(0); // indicate we have yet to complete this phase by resetting progress
+ } else {
+ _state.set_completed();
+ }
+ } else {
Needed::buf_t tmp;
it->buf.swap (tmp); // deallocates the space...
it->nntp = 0;
@@ -271,130 +280,108 @@
check_in (nntp, health);
}
-namespace
+void
+TaskArticle :: on_work_cancelled(void *data)
{
- void uu_log (void* unused, char* message, int severity)
- {
- char * pch (g_locale_to_utf8 (message, -1, 0, 0, 0));
+ (void)data;
- 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);
+ assert(_decoder);
+ if (!_decoder) return;
- g_free (pch);
- }
+ _state.set_need_decode();
+ set_step(0);
+ Decoder::check_in(_decoder);
+ _decoder = 0;
}
+
+// this is called as per the WorkerPool::Listener interface and it means that the save thread finished
void
-TaskArticle :: on_finished ()
+TaskArticle :: on_work_complete(void *data)
{
- const Article::mid_sequence_t mids (_article.get_part_mids());
- const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+ (void) data;
- 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);
- }
- }
+ 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
+ bool have_errs = false;
- 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));
- }
-
- 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());
-
- // 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);
- }
-
- // 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));
- }
+ // when we get here, our decoder thread finished.. so pick up the logs it
+ // generated and put them in the log here in the main thread, and also
+ // update the _read object to mark the article read, and also
+ // notify listeners that task_decode stuff is completed
+ foreach_const(std::list<std::string>, _decoder->log_urgents, it)
+ Log :: add_urgent(it->c_str()), have_errs = true;
+ foreach_const(std::list<std::string>, _decoder->log_errors, it)
+ Log :: add_err(it->c_str()), have_errs = true;
+ foreach_const(std::list<std::string>, _decoder->log_infos, it)
+ Log :: add_info(it->c_str());
+
+ if (_decoder->mark_read) _read.mark_read(_article);
+
+ _state.set_completed(); // mark us COMPLETED here..
+ set_step(100); // indicate we're done to interested parties
+
+ if (have_errs) {
+ const std::string &s =
+ !_decoder->log_errors.empty()
+ ? _decoder->log_errors.front()
+ : _decoder->log_urgents.front();
+ fire_task_decode_error(s);
+ }
+
+ Decoder::check_in(_decoder);
+ _decoder = 0;
+
+ // notify interested object that the saver thread is free and so they can try to save.. for now just the Queue is a listener
+ fire_task_finished_decoding();
+ set_step(100);
+ update_work();
+ debug("decoder thread done");
+}
- // cleanup
- g_free (fname);
- }
+bool TaskArticle :: try_decode()
+{
+ bool ret = false;
+ assert(_state._work == NEED_DECODE);
- _read.mark_read (_article);
+ if (_state._work == NEED_DECODE) {
+ if ( (_decoder = Decoder::check_out()) ) { // is thread free?
+ // yay thread is free
+ const Article::mid_sequence_t mids (_article.get_part_mids());
+ const ArticleCache :: strings_t filenames (_cache.get_filenames (mids));
+
+ init_steps(100);
+ _state.set_decoding();
+
+ _decoder->enqueue_work_in_thread(this, _decoder, _save_path, filenames, _save_mode);
+
+ set_status_va (_("Decoding %s"), _article.subject.c_str());
+ //increment_step();
+ ret = true;
+ debug("decoder thread was free, enqueued work");
+ } else {
+ debug("decoder thread was busy, we'll try later");
}
- UUCleanUp ();
}
+ return ret;
+}
+
+void
+TaskArticle :: decode_cancel ()
+{
+ if (_state._work == DECODING && _decoder) {
+ debug("Decoder cancellation request -- forwarding to decoder object");
+ _decoder->cancel();
+ }
+ // will set the cancel state when our on_work_cancelled callback finally runs and the decode ends
+}
+
+void
+TaskArticle :: on_finished ()
+{
+ // noop -- used to uudecode and/or save attachements here, but this was
+ // moved to the Decoder object above.. see task-article-decoder.[ch]* and try_decode() above..
}
unsigned long
@@ -405,3 +392,11 @@
bytes += (it->part.bytes - it->buf.size());
return bytes;
}
+
+
+TaskArticle :: ~TaskArticle ()
+{
+ decode_cancel(); // cancel running decode, in order to release thread asap
+ _decoder = 0;
+ _cache.release (_article.get_part_mids());
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task-article.h pan2.svn-decoder-threads-works!/pan/tasks/task-article.h
--- pan2.svn-orig/pan/tasks/task-article.h 2007-03-20 06:17:58.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task-article.h 2007-03-19 21:20:02.000000000 -0400
@@ -27,14 +27,18 @@
#include <pan/data/xref.h>
#include <pan/tasks/nntp.h>
#include <pan/tasks/task.h>
+#include <pan/tasks/worker-pool.h>
namespace pan
{
+
/**
* 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::Listener
{
public: // life cycle
@@ -45,16 +49,25 @@
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;
+
+ /** 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.
+ */
+ bool try_decode();
+ void decode_cancel();
private: // Task subclass
virtual void use_nntp (NNTP * nntp);
@@ -63,6 +76,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;
@@ -78,6 +95,10 @@
const SaveMode _save_mode;
typedef std::map<Quark,int> stats_t;
stats_t _stats;
+
+ class Decoder; // see task-article-decoder.h
+ friend class Decoder;
+ Decoder *_decoder;
private:
struct Needed {
@@ -93,6 +114,7 @@
needed_t _needed;
void update_work ();
+
};
}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task.cc pan2.svn-decoder-threads-works!/pan/tasks/task.cc
--- pan2.svn-orig/pan/tasks/task.cc 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task.cc 2007-03-19 20:14:33.000000000 -0400
@@ -20,6 +20,7 @@
#include <config.h>
#include <pan/general/debug.h>
#include <pan/general/messages.h>
+#include <pan/general/foreach.h>
#include "task.h"
using namespace pan;
@@ -75,3 +76,17 @@
return oldest_time;
}
#endif
+
+void Task::fire_task_finished_decoding () {
+ foreach (listeners_t, _listeners, it) {
+ Task::Listener * l = dynamic_cast<Task::Listener *>(*it);
+ if (l) l->on_task_finished_decoding (this);
+ }
+}
+
+void Task::fire_task_decode_error (const StringView &message) {
+ foreach (listeners_t, _listeners, it) {
+ Task::Listener * l = dynamic_cast<Task::Listener *>(*it);
+ if (l) l->on_task_decode_error (this, message);
+ }
+}
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/task.h pan2.svn-decoder-threads-works!/pan/tasks/task.h
--- pan2.svn-orig/pan/tasks/task.h 2007-03-20 06:17:57.000000000 -0400
+++ pan2.svn-decoder-threads-works!/pan/tasks/task.h 2007-03-19 23:05:31.000000000 -0400
@@ -51,7 +51,11 @@
/** Task is waiting on an nntp connection */
NEED_NNTP,
/** Task is running */
- WORKING
+ WORKING,
+ /** Task waiting for decoder thread */
+ NEED_DECODE,
+ /** Task saving */
+ DECODING,
};
/**
@@ -85,6 +89,12 @@
void set_need_nntp (const Quark& server) {
_work=NEED_NNTP; _servers.clear(); _servers.insert(server); }
+ void set_need_decode () {
+ _work = NEED_DECODE; _servers.clear(); }
+
+ void set_decoding () {
+ _work = DECODING; _servers.clear(); }
+
void set_health (Health h) {
_health = h; }
@@ -108,6 +118,17 @@
const Quark& get_type () const { return _type; }
virtual unsigned long get_bytes_remaining () const = 0;
+
+ // for now, just used to inform the queue about task states changing
+ struct Listener : public Progress::Listener {
+ virtual void on_task_finished_decoding(Task *) = 0;
+ virtual void on_task_decode_error(Task *, const StringView &msg) = 0;
+ };
+
+ /// see TaskArticle for a class that actually uses this method
+ virtual bool try_decode() { return false; }
+ /// cancel an already-running decode -- TaskArticle actually uses this
+ virtual void decode_cancel() { }
protected:
@@ -119,6 +140,9 @@
int get_nntp_count () const { return _nntp_to_source.size(); }
+ void fire_task_finished_decoding();
+ void fire_task_decode_error(const StringView & message);
+
private:
/** What type this task is ("XOVER", "POST", "SAVE", "BODIES", etc...) */
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/worker-pool.cc pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.cc
--- pan2.svn-orig/pan/tasks/worker-pool.cc 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.cc 2007-03-19 21:55:27.000000000 -0400
@@ -0,0 +1,143 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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 <pan/general/debug.h>
+#include <pan/general/foreach.h>
+#include "worker-pool.h"
+#include <cassert>
+#include <set>
+#include <map>
+
+namespace pan
+{
+
+ namespace {
+ WorkerPool::WorkerSet all_workers;
+ }
+
+ WorkerPool::Worker::Worker() : please_stop(false)
+ {
+ all_workers.insert(this);
+ }
+
+ WorkerPool::Worker::~Worker()
+ {
+ all_workers.erase(this);
+ }
+
+ void
+ WorkerPool::cancelAllWorkers()
+ {
+ debug("notifying all workers to stop");
+ foreach(WorkerSet, all_workers, it)
+ (*it)->cancel();
+ }
+
+ struct WorkerPool::Work
+ {
+ Work() : worker(0), listener(0), delete_worker(false), data(0) {}
+ Worker *worker;
+ WorkerPool *pool;
+ Listener *listener;
+ bool delete_worker;
+ void *data;
+ };
+
+ WorkerPool::WorkerPool(int nthr, bool exclusive)
+ {
+ if (!g_thread_supported ()) g_thread_init (NULL);
+ //g_thread_pool_set_max_idle_time(1000); //1s idle time enough to reap idle threads? if users of this class care, perhaps make this a setable global property..
+ assert(!exclusive || nthr > -1); // assert that for exclusive threads, nthr cannot be negative
+ tpool = g_thread_pool_new(thr_wrapper,
+ reinterpret_cast<gpointer>(this),
+ nthr,
+ exclusive, /* exclusive? */
+ 0);
+ }
+
+ WorkerPool::~WorkerPool()
+ {
+ debug("Deleting thread pool 0x" << ((void *)this) << " with max threads: " << maxThreads() << " num threads: " << numThreads() << " unprocessed: " << unprocessed());
+
+ // ask my_workers to die
+ foreach(WorkerSet, my_workers, it)
+ (*it)->cancel();
+
+ g_thread_pool_free(tpool, false, true);
+ }
+
+ /* 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->listener) {
+ if (work->worker->was_cancelled())
+ work->listener->on_work_cancelled(work->data);
+ else
+ work->listener->on_work_complete(work->data);
+ }
+ work->pool->my_workers.erase(work->worker);
+ 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, Listener *listener, bool delworker)
+ {
+ Work *w = new Work;
+ w->worker = worker;
+ w->listener = listener;
+ w->delete_worker = delworker;
+ w->data = data;
+ w->pool = this;
+ my_workers.insert(worker);
+ g_thread_pool_push(tpool, w, NULL); // ends up invokint thr_wrapper in thread
+ }
+
+
+ void WorkerPool::setMaxThreads(int num) {
+ g_thread_pool_set_max_threads(tpool, num, NULL);
+ }
+ int WorkerPool::maxThreads(void) const {
+ return g_thread_pool_get_max_threads(tpool);
+ }
+ unsigned WorkerPool::numThreads(void) const {
+ return g_thread_pool_get_num_threads(tpool);
+ }
+ unsigned WorkerPool::unprocessed(void) const {
+ return g_thread_pool_unprocessed(tpool);
+ }
+
+}
+
diff --exclude-from=diff-excludes -urN pan2.svn-orig/pan/tasks/worker-pool.h pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.h
--- pan2.svn-orig/pan/tasks/worker-pool.h 1969-12-31 19:00:00.000000000 -0500
+++ pan2.svn-decoder-threads-works!/pan/tasks/worker-pool.h 2007-03-19 21:53:53.000000000 -0400
@@ -0,0 +1,107 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/*
+ * Pan - A Newsreader for Gtk+
+ * Copyright (C) 2007 Calin Culianu <[email protected]>
+ *
+ * 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.
+ *
+ * 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.h>
+#include <set>
+
+namespace pan
+{
+ /**
+ * This class basically encapsulates creation and management
+ * of the pan-specific worker-pool mechanism. The task queue in pan
+ * makes use of this class. */
+ class WorkerPool
+ {
+ public:
+ WorkerPool(int num_threads = -1, bool exclusive = true); ///< 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(); ///< deletes all the threads, may or may not stop workers
+
+ /// call this on app exit to try and notify all workers to stop!
+ static void cancelAllWorkers();
+
+ /** set the limit on number of threads -- semantics just like the glib
+ funtion */
+ void setMaxThreads(int);
+ int maxThreads(void) const;
+
+ /// returns number of threads currently runing
+ unsigned numThreads(void) const;
+ /// returns the number of tasks still unprocess in pool
+ unsigned unprocessed(void) const;
+
+ 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) {}
+ };
+
+ 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; }
+ virtual bool was_cancelled() { return please_stop; }
+
+ 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;
+ 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, Listener * = 0, bool delete_worker_on_completion = false);
+
+
+ typedef std::set<Worker *> WorkerSet;
+
+ 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;
+ WorkerSet my_workers;
+
+ };
+
+};
+
+#endif