schedular.c

Go to the documentation of this file.

/*
 Copyright (C) 2005-2007 Erik van Pienbroek

 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; 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 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#ifdef WIN32
#include 
#endif
#include "nntpgrab_plugin_schedular.h"
#include "schedular_plugin.h"

typedef struct SchedularPluginClass SchedularPluginClass;

static SchedularPlugin *schedular;

struct SchedularPluginClass
{
    GObjectClass parent;
};

G_DEFINE_TYPE(SchedularPlugin, schedular_plugin, G_TYPE_OBJECT);

gboolean
nntpgrab_plugin_schedular_initialize(PluginSchedularImportedFuncs funcs, char **errmsg)
{
    schedular_plugin_set_imported_funcs(funcs);

    schedular = g_object_new(SCHEDULAR_PLUGIN_TYPE_OBJECT, NULL);

    if (!load_download_queue(&schedular->queue, errmsg)) {
        g_object_unref(schedular);

        return FALSE;
    }

    return TRUE;
}

void
nntpgrab_plugin_schedular_destroy(void)
{
    g_object_unref(schedular);
}

int
nntpgrab_plugin_schedular_get_version(void)
{
    return NNTPGRAB_PLUGIN_API_VERSION;
}

static void
schedular_plugin_init (SchedularPlugin *obj)
{
    g_static_mutex_init(&obj->mutex);
    g_static_mutex_init(&obj->runtime_mutex);

    obj->thread = NULL;
    obj->abort_flag = FALSE;
    obj->queue = NULL;
}

static void
schedular_plugin_finalize (GObject *obj)
{
    SchedularPlugin *schedular = SCHEDULAR_PLUGIN(obj);
    GList *list;

    g_static_mutex_lock(&schedular->runtime_mutex);

    if (schedular->thread) {
        g_static_mutex_lock(&schedular->mutex);
        schedular->abort_flag = TRUE;
        g_static_mutex_unlock(&schedular->mutex);

        g_thread_join(schedular->thread);
    }

    g_static_mutex_unlock(&schedular->runtime_mutex);

    g_static_mutex_lock(&schedular->mutex);

    // Clear all the configured usenet servers
    list = schedular->servers;
    while (list) {
        g_slice_free(ConfigServer, list->data);
        list = g_list_next(list);
    }
    g_list_free(schedular->servers);
    schedular->servers = NULL;

    save_download_queue(schedular->queue, NULL);        // TODO: Error message not caught

    free_download_queue(schedular->queue);

    g_static_mutex_unlock(&schedular->mutex);

    g_static_mutex_free(&schedular->mutex);
    g_static_mutex_free(&schedular->runtime_mutex);
}

static void
schedular_plugin_class_init (SchedularPluginClass *klass)
{
    GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

    gobject_class->finalize = schedular_plugin_finalize;
}

struct _part_download_done {
    NNTPCollection *collection;
    NNTPFile *file;
    NNTPPart *part;
};

gboolean
get_next_part_to_download(SchedularPlugin *schedular, int server_id, NNTPCollection **collection, NNTPFile **file, NNTPPart **part, GThreadPool *poolDecoder)
{
    struct _part_download_done *job;

    job = g_async_queue_try_pop(schedular->task_queue[server_id]);
    if (!job) {
        //g_mutex_lock(schedular->task_mutex);
        g_cond_signal(schedular->task_cond);
        //g_mutex_unlock(schedular->task_mutex);

        job = g_async_queue_try_pop(schedular->task_queue[server_id]);

        if (!job) {
            return FALSE;
        }
    }

    *collection = job->collection;
    *file = job->file;
    *part = job->part;

    g_slice_free(struct _part_download_done, job);

    // Is this collection being removed ?
    if ((*collection)->stop_flag || (*file)->stop_flag) {
        update_part_download_status(*collection, *file, *part, server_id, FALSE, TRUE, FALSE, FALSE, NULL);
        return FALSE;
    }

    //g_mutex_lock(schedular->task_mutex);
    g_cond_signal(schedular->task_cond);
    //g_mutex_unlock(schedular->task_mutex);

    return TRUE;
}

static gboolean
test_server_may_download_part(NNTPPart *part, int server_id)
{
    int i;
    gboolean all_tried;

    if (!schedular->is_server_enabled[server_id]) {
        return FALSE;
    }

    if (part->servers_already_tried[server_id]) {
        return FALSE;
    }

    // Is this server one with the highest priority ?
    if (schedular->is_high_priority_server[server_id]) {
        return TRUE;
    }

    all_tried = TRUE;

    // Are all the other high-priority servers already tried ?
    for (i = 0; i MAX_NNTP_SERVERS; i++) {
        if (schedular->is_high_priority_server[i] && !part->servers_already_tried[i]) {
            all_tried = FALSE;
        }
    }

    if (!all_tried) {
        return FALSE;
    }

    // All the high-priority servers are already tried, now test against the servers with the 'normal' priority
    if (schedular->is_normal_priority_server[server_id]) {
        return TRUE;
    }

    all_tried = TRUE;
    for (i = 0; i 00213         if (schedular->is_normal_priority_server[i] && !part->servers_already_tried[i]) {
            all_tried = FALSE;
        }
    }

    if (!all_tried) {
        return FALSE;
    }

    // All the normal priority servers are also tried, allow the part

    return TRUE;
}

static gboolean
get_next_job_to_schedule(SchedularPlugin *schedular, int server_id, NNTPCollection **collection, NNTPFile **file, NNTPPart **part, GThreadPool *poolDecoder)
{
    GList *list;

    g_static_mutex_lock(&schedular->mutex);

    list = schedular->queue;
    while (list) {
        GList *list2;

        *collection = (NNTPCollection *) list->data;

        // Is this collection being removed ?
        if ((*collection)->stop_flag) {
            list = g_list_next(list);
            continue;
        }

        list2 = (*collection)->files;
        while (list2) {
            GList *list3;
            DecoderData *decoder_data;
            gboolean complete_flag;

            *file = (NNTPFile *) list2->data;

            // Is this file already downloaded?
            if ((*file)->file_is_downloaded) {
                list2 = g_list_next(list2);
                continue;
            }

            if ((*file)->now_decoding) {
                list2 = g_list_next(list2);
                continue;
            }

            // Is this file being removed?
            if ((*file)->stop_flag) {
                list2 = g_list_next(list2);
                continue;
            }

            complete_flag = TRUE;
            list3 = (*file)->parts;
            while (list3) {
                *part = (NNTPPart *) list3->data;

                if ((*part)->now_downloading) {
                    complete_flag = FALSE;
                }

                if (!(*part)->downloaded) {
                    int i;
                    gboolean all_servers_tried;

                                        if (!(*part)->now_downloading && test_server_may_download_part(*part, server_id)) {
                        (*part)->now_downloading = TRUE;
                        (*file)->file_is_downloading = TRUE;
                        g_static_mutex_unlock(&schedular->mutex);

                        return TRUE;
                    }

                    // Are all the servers already tried ?
                    all_servers_tried = TRUE;

                    for (i = 0; i servers); i++) {
                        if ((!(*part)->servers_already_tried[i]) && schedular->is_server_enabled[i]) {
                            all_servers_tried = FALSE;
                        }
                    }

                    if (!all_servers_tried) {
                        complete_flag = FALSE;
                    }
                }

                list3 = g_list_next(list3);
            }

            // All the parts are already downloaded?
            if (complete_flag && !(*file)->now_decoding) {
                (*file)->now_decoding = TRUE;

                imported_funcs_task_state_changed((*collection)->collection_name, (*file)->subject, (*file)->status, TASK_STATE_WAITING_FOR_DECODE);

                (*file)->status = TASK_STATE_WAITING_FOR_DECODE;

                decoder_data = g_slice_new0(DecoderData);
                decoder_data->collection_name = (*collection)->collection_name;
                decoder_data->file = (*file);
                decoder_data->schedular = schedular;

                g_thread_pool_push(poolDecoder, decoder_data, NULL);
            }

            list2 = g_list_next(list2);
        }

        list = g_list_next(list);
    }

    g_static_mutex_unlock(&schedular->mutex);

    *collection = NULL;
    *file = NULL;
    *part = NULL;

    return FALSE;
}

static int
get_max_threads_for_server(SchedularPlugin *schedular, int server_id)
{
    GList *list;
    ConfigServer *server;
    int ret;

    g_static_mutex_lock(&schedular->mutex);
    list = g_list_nth(schedular->servers, server_id);

    server = list->data;

    g_return_val_if_fail(server != NULL, -1);

    ret = server->max_threads;

    g_static_mutex_unlock(&schedular->mutex);

    return ret;
}

static void
prepare_jobs(SchedularPlugin *schedular, GThreadPool *poolDecoder)
{
    int i;
    struct _part_download_done *job;

    for (i = 0; i 00368         while (schedular->task_queue[i] && g_async_queue_length(schedular->task_queue[i]) 00369             // Push a new job into this queue
            NNTPCollection *collection = NULL;
            NNTPFile *file = NULL;
            NNTPPart *part = NULL;

            if (!get_next_job_to_schedule(schedular, i, &collection, &file, &part, poolDecoder)) {
                // Nothing to schedule
                break;
            }

            job = g_slice_new0(struct _part_download_done);
            job->collection = collection;
            job->file = file;
            job->part = part;

            g_async_queue_push(schedular->task_queue[i], job);
        };
    }
}

void
update_part_download_status(NNTPCollection *collection, NNTPFile *file, NNTPPart *part, int server_id, gboolean success, gboolean reset_to_zero, gboolean queue_is_being_paused, gboolean dont_retry_anymore, gpointer nntpconnection)
{
    GList *list;
    ConfigServer *server;

    g_static_mutex_lock(&schedular->mutex);

    part->now_downloading = FALSE;
    part->downloaded = success;
    file->file_is_downloading = FALSE;

    if (success) {
        if (file->file_size_remaining size) {
            g_print("Invalid file size remaining!\n");
#if SIZEOF_LONG == 8
            g_print("file->file_size_remaining = %li\n", file->file_size_remaining);
#else
            g_print("file->file_size_remaining = %lli\n", file->file_size_remaining);
#endif
            g_print("part->size = %i\n", part->size);
            file->file_size_remaining = part->size;
        }

        if (collection->total_size_remaining size) {
            g_print("Invalid collection size remaining!\n");
#if SIZEOF_LONG == 8
            g_print("collection->total_size_remaining = %li\n", collection->total_size_remaining);
#else
            g_print("collection->total_size_remaining = %lli\n", collection->total_size_remaining);
#endif
            g_print("part->size = %i\n", part->size);
            collection->total_size_remaining = part->size;
        }

        file->file_size_remaining -= part->size;
        collection->total_size_remaining -= part->size;
        file->num_parts_downloaded++;

        server = g_list_nth_data(schedular->servers, server_id);
        imported_funcs_emit_part_download_complete(server->servername, nntpconnection, collection->collection_name, file->subject, part->partnum, part->size);
        imported_funcs_emit_file_download_state_update(collection->collection_name, file->subject, file->numparts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);
    } else if (!success && dont_retry_anymore) {
        part->servers_already_tried[server_id] = TRUE;
    }

    // Are other threads downloading parts of this file ?
    list = file->parts;
    while (list) {
        NNTPPart *part2 = list->data;

        if (part2->now_downloading) {
            file->file_is_downloading = TRUE;
        }

        list = g_list_next(list);
    }

    // Are all the servers already tried ?
    if (!success && dont_retry_anymore) {
        int i;
        gboolean all_servers_tried = TRUE;

        for (i = 0; i servers); i++) {
            if (!part->servers_already_tried[i] && schedular->is_server_enabled[i]) {
                all_servers_tried = FALSE;
            }
        }

        server = g_list_nth_data(schedular->servers, server_id);

        if (nntpconnection) {
            imported_funcs_emit_part_download_failed (server->servername, nntpconnection, collection->collection_name, file->subject, part->partnum, part->size, all_servers_tried);
        }

        if (all_servers_tried && !reset_to_zero) {
            if (file->file_size_remaining size) {
                g_print("Invalid file size remaining!\n");
#if SIZEOF_LONG == 8
                g_print("file->file_size_remaining = %li\n", file->file_size_remaining);
#else
                g_print("file->file_size_remaining = %lli\n", file->file_size_remaining);
#endif
                g_print("part->size = %i\n", part->size);
                file->file_size_remaining = part->size;
            }

            if (collection->total_size_remaining size) {
                g_print("Invalid collection size remaining!\n");
#if SIZEOF_LONG == 8
                g_print("collection->total_size_remaining = %li\n", collection->total_size_remaining);
#else
                g_print("collection->total_size_remaining = %lli\n", collection->total_size_remaining);
#endif
                g_print("part->size = %i\n", part->size);
                collection->total_size_remaining = part->size;
            }

            file->num_parts_failed++;
            file->file_size_remaining -= part->size;
            collection->total_size_remaining -= part->size;

            imported_funcs_emit_file_download_state_update(collection->collection_name, file->subject, file->numparts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);
        }
    }

    if (reset_to_zero) {
        file->num_parts_downloaded = 0;
        file->num_parts_failed = 0;

        imported_funcs_emit_file_download_state_update(collection->collection_name, file->subject, file->numparts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);
    }

    g_static_mutex_unlock(&schedular->mutex);
}

static gpointer
stop_schedular_func(gpointer data)
{
    nntpgrab_plugin_schedular_stop((const char *) data);

    g_free(data);

    return NULL;
}

void
stop_schedular_from_seperate_thread(const char *reason)
{
    // The schedular needs to be stopped from a seperate thread in order to prevent deadlocks
    g_thread_create(stop_schedular_func, g_strdup(reason), FALSE, NULL);
}

static gpointer
schedular_thread_func(gpointer data)
{
    SchedularPlugin *schedular = SCHEDULAR_PLUGIN(data);
    GThreadPool *pool[MAX_NNTP_SERVERS];
    GThreadPool *poolDecoder;
    GList *list;
    int i = 0;
    gboolean abort_flag;
    GTimeVal val;

    // Initialize a thread pool for all the configured servers
    g_assert(g_list_length(schedular->servers) 00535     memset(pool, 0, sizeof(pool));

    poolDecoder = g_thread_pool_new(decoder_thread_func, schedular, 1, FALSE, NULL);

    schedular->task_cond = g_cond_new();
    schedular->task_mutex = g_mutex_new();
    memset(schedular->task_queue, 0, sizeof(schedular->task_queue));

    list = schedular->servers;
    while (list) {
        ConfigServer *server = list->data;
        int j;

        schedular->task_queue[i] = g_async_queue_new();

        // If this server enabled?
        if (!schedular->is_server_enabled[i]) {
            pool[i] = NULL;
            i++;
            list = g_list_next(list);
            continue;
        }

        pool[i] = g_thread_pool_new(download_thread_func, schedular, server->max_threads, FALSE, NULL);

        g_assert(pool[i]);

        // Launch all threads in the pool
        for (j = 0; j max_threads; j++) {
            DownloadData *download_data = g_slice_new(DownloadData);
            download_data->server = server;
            download_data->server_id = i;
            download_data->poolDecoder = poolDecoder;
            g_thread_pool_push(pool[i], download_data, NULL);
        }

        i++;

        list = g_list_next(list);
    }

    // Create a initial list of jobs
    prepare_jobs(schedular, poolDecoder);

    abort_flag = FALSE;
    while (!abort_flag) {
        g_static_mutex_lock(&schedular->mutex);
        abort_flag = schedular->abort_flag;
        g_static_mutex_unlock(&schedular->mutex);

        g_get_current_time(&val);
        g_time_val_add(&val, G_USEC_PER_SEC);

        g_mutex_lock(schedular->task_mutex);

        if (g_cond_timed_wait(schedular->task_cond, schedular->task_mutex, &val)) {
            // One or more threads want new data
            prepare_jobs(schedular, poolDecoder);
        }

        g_mutex_unlock(schedular->task_mutex);
    }

    for (i = 0; i 00599         if (pool[i]) {
            g_thread_pool_free(pool[i], TRUE, TRUE);
        }

        if (schedular->task_queue[i]) {
            // Free any remaining jobs
            struct _part_download_done *job;
            while ((job = g_async_queue_try_pop(schedular->task_queue[i]))) {
                // Switch the 'now_downloading' status to FALSE
                update_part_download_status(job->collection, job->file, job->part, i, FALSE, FALSE, TRUE, FALSE, NULL);

                g_slice_free(struct _part_download_done, job);
            }

            g_async_queue_unref(schedular->task_queue[i]);
            schedular->task_queue[i] = NULL;
        }
    }

    g_thread_pool_free(poolDecoder, FALSE, TRUE);
    g_cond_free(schedular->task_cond);
    g_mutex_free(schedular->task_mutex);

    return NULL;
}

gboolean
nntpgrab_plugin_schedular_start(void)
{
    GError *error = NULL;
    GList *list;
    GList *servers;
    int i;

    g_assert(schedular != NULL);

    g_static_mutex_lock(&schedular->mutex);

    if (schedular->thread != NULL) {
        g_static_mutex_unlock(&schedular->mutex);
        return FALSE;
    }

    // Make a copy of the servers list
    g_assert(schedular->servers == NULL);

    memset(&schedular->is_high_priority_server, 0, sizeof(schedular->is_high_priority_server));
    memset(&schedular->is_normal_priority_server, 0, sizeof(schedular->is_normal_priority_server));
    memset(&schedular->is_low_priority_server, 0, sizeof(schedular->is_low_priority_server));
    memset(&schedular->is_server_enabled, 0, sizeof(schedular->is_server_enabled));

    servers = imported_funcs_config_get_avail_servers();
    list = servers;
    i = 0;

    while (list) {
        char *servername = list->data;
        ConfigServer *server = imported_funcs_config_get_server_info(servername);

        g_assert(server);

        schedular->servers = g_list_append(schedular->servers, server);
        if (server->enabled) {
            if (server->priority == SERVER_PRIORITY_HIGH) {
                schedular->is_high_priority_server[i] = TRUE;
                schedular->is_normal_priority_server[i] = FALSE;
                schedular->is_low_priority_server[i] = FALSE;
            } else if (server->priority == SERVER_PRIORITY_LOW) {
                schedular->is_high_priority_server[i] = FALSE;
                schedular->is_normal_priority_server[i] = FALSE;
                schedular->is_low_priority_server[i] = TRUE;
            } else {
                schedular->is_high_priority_server[i] = FALSE;
                schedular->is_normal_priority_server[i] = TRUE;
                schedular->is_low_priority_server[i] = FALSE;
            }
        }

        schedular->is_server_enabled[i] = server->enabled;

        i++;

        g_free(servername);

        list = g_list_next(list);
    }

    g_list_free(servers);

    // Reset the abort flag
    schedular->abort_flag = FALSE;

    // Initialize and start the real schedular thread
    if (!(schedular->thread = g_thread_create(schedular_thread_func, schedular, TRUE, &error))) {
        imported_funcs_emit_fatal_error(error->message);
        g_error_free(error);
    }

    g_static_mutex_unlock(&schedular->mutex);

    imported_funcs_notify_schedular_state_changed(SCHEDULAR_STATE_RUNNING, NULL);

    return TRUE;
}

gboolean
nntpgrab_plugin_schedular_stop(const char *reason)
{
    GList *list;

    g_assert(schedular != NULL);

    g_static_mutex_lock(&schedular->runtime_mutex);

    if (schedular->thread == NULL) {
        g_static_mutex_unlock(&schedular->runtime_mutex);
        return FALSE;
    }

    imported_funcs_notify_schedular_state_changed(SCHEDULAR_STATE_STOPPING, reason);

    // Abort the schedular thread
    g_static_mutex_lock(&schedular->mutex);
    schedular->abort_flag = TRUE;
    g_static_mutex_unlock(&schedular->mutex);

    g_thread_join(schedular->thread);
    schedular->thread = NULL;

    // Clear all the configured usenet servers
    g_static_mutex_lock(&schedular->mutex);
    list = schedular->servers;
    while (list) {
        g_slice_free(ConfigServer, list->data);
        list = g_list_next(list);
    }
    g_list_free(schedular->servers);
    schedular->servers = NULL;
    g_static_mutex_unlock(&schedular->mutex);

    g_static_mutex_unlock(&schedular->runtime_mutex);

    imported_funcs_notify_schedular_state_changed(SCHEDULAR_STATE_STOPPED, NULL);

    return TRUE;
}

SchedularState
nntpgrab_plugin_schedular_get_state(void)
{
    SchedularState state;

    g_static_mutex_lock(&schedular->mutex);
    if (schedular->thread && schedular->abort_flag) {
        state = SCHEDULAR_STATE_STOPPING;
    } else if (schedular->thread && !schedular->abort_flag) {
        state = SCHEDULAR_STATE_RUNNING;
    } else {
        state = SCHEDULAR_STATE_STOPPED;
    }
    g_static_mutex_unlock(&schedular->mutex);

    return state;
}

static gboolean
test_is_subject_already_in_collection(NNTPCollection *collection, const char *subject)
{
    GList *list;

    list = collection->files;
    while (list) {
        NNTPFile *file = (NNTPFile *) list->data;

        if (!strcmp(file->subject, subject)) {
            return TRUE;
        }

        list = g_list_next(list);
    }

    return FALSE;
}

gboolean
nntpgrab_plugin_schedular_add_task_to_queue(const char *collection_name, NNTPFile *file, char **errmsg)
{
    GList *list;
    NNTPCollection *collection;

    g_static_mutex_lock(&schedular->mutex);

    // Find out if the collection_name already exists in the download queue
    list = schedular->queue;
    while (list) {
        collection = (NNTPCollection *) list->data;

        if (!strcmp(collection->collection_name, collection_name)) {
            // Collection found!

            // Does the subject already exists in this collection ?
            if (test_is_subject_already_in_collection(collection, file->subject)) {
                if (errmsg) {
                    *errmsg = g_strdup_printf(_("There already is a file with subject '%s' part of collection '%s'"), file->subject, collection_name);
                }

                g_static_mutex_unlock(&schedular->mutex);

                return FALSE;
            }

            // Add the file to this collection
            collection->files = g_list_append(collection->files, file);
            collection->total_size += file->file_size;
            collection->total_size_remaining += file->file_size;
            file->file_size_remaining = file->file_size;

            file->position = g_list_index(collection->files, file) + 1;

            imported_funcs_task_added(collection, file);

            g_static_mutex_unlock(&schedular->mutex);

            return TRUE;
        }

        list = g_list_next(list);
    }

    // Collection doesn't exist yet, add a new one
    collection = g_slice_new0(NNTPCollection);
    strncpy((char *) collection->collection_name, collection_name, sizeof(collection->collection_name));
    schedular->queue = g_list_append(schedular->queue, collection);

    imported_funcs_collection_added(collection_name);

    // Finally add the task to the collection
    collection->files = g_list_append(collection->files, file);
    collection->total_size = file->file_size;
    collection->total_size_remaining = file->file_size;
    file->position = 1;

    imported_funcs_task_added(collection, file);

    g_static_mutex_unlock(&schedular->mutex);

    return TRUE;
}

static void
free_file(NNTPFile *file)
{
    GList *list;

    g_assert(file);

    list = file->parts;
    while (list) {
        g_slice_free(NNTPPart, list->data);
        list = g_list_next(list);
    }
    g_list_free(file->parts);
    file->parts = NULL;

    list = file->groups;
    while (list) {
        g_free(list->data);
        list = g_list_next(list);
    }
    g_list_free(file->groups);
    file->groups = NULL;

    g_free(file);
}

gboolean
nntpgrab_plugin_schedular_del_task_from_queue(const char *collection_name, const char *subject, char **errmsg)
{
    GList *list;
    NNTPCollection *collection;
    int i;
    ConfigOpts opts;

    g_static_mutex_lock(&schedular->mutex);

    opts = imported_funcs_config_get_opts();

    // Find out if the collection_name exists in the download queue
    list = schedular->queue;
    while (list) {
        collection = (NNTPCollection *) list->data;

        if (!strcmp(collection->collection_name, collection_name)) {
            // Collection found!
            GList *list2;
            gboolean delete_flag = FALSE;

            // Do we have a subject?
            if (!subject) {
                // Delete the whole collection
                delete_flag = TRUE;

                // Mark the whole collection as being removed
                collection->stop_flag = TRUE;
            }

            // Try to find the subject
            list2 = collection->files;
            while (list2) {
                NNTPFile *file = (NNTPFile *) list2->data;

                list2 = g_list_next(list2);

                if (delete_flag || !strcmp(file->subject, subject)) {
                    // Bingo!
                    // Is this file being downloaded ?
                    while (file->file_is_downloading || file->now_decoding) {
                        // Mark this file as being removed
                        file->stop_flag = TRUE;

                        // Release the lock and wait for the download to complete
                        g_static_mutex_unlock(&schedular->mutex);

                        g_usleep(G_USEC_PER_SEC / 10);

                        g_static_mutex_lock(&schedular->mutex);
                    }

                    // Unlink any possible downloaded parts
                    for (i = 0; i numparts; i++) {
                        char *filename = g_strdup_printf("%s%s%s.%i", opts.temp_directory, G_DIR_SEPARATOR_S, file->tmp_filename, i + 1);
                        g_unlink(filename);
                        g_free(filename);
                    }

                    // Update the file size
                    collection->total_size -= file->file_size;
                    collection->total_size_remaining -= file->file_size_remaining;

                    // Remove it from the list
                    collection->files = g_list_remove(collection->files, file);

                    // Restart from the beginning of the list as we have removed an item
                    list2 = collection->files;

                    imported_funcs_task_removed(collection_name, file->subject, collection->total_size, collection->total_size_remaining);

                    free_file(file);

                    // Does the collection still got any items in it ?
                    if (g_list_length(collection->files) == 0) {
                        // Yes, remove the entire collection
                        schedular->queue = g_list_remove(schedular->queue, collection);

                        imported_funcs_collection_removed(collection_name);

                        g_slice_free(NNTPCollection, collection);
                    }

                    if (!delete_flag) {
                        g_static_mutex_unlock(&schedular->mutex);
                        return TRUE;
                    }
                }
            }

            if (delete_flag) {
                g_static_mutex_unlock(&schedular->mutex);
                return TRUE;
            }
        }

        list = g_list_next(list);
    }

    if (errmsg) {
        *errmsg = g_strdup_printf(_("Subject '%s' was not found in collection '%s'"), subject, collection_name);
    }

    g_static_mutex_unlock(&schedular->mutex);

    return FALSE;
}

gboolean
nntpgrab_plugin_schedular_restart_task(const char *collection_name, const char *subject, char **errmsg)
{
    GList *list;
    NNTPCollection *collection;

    g_static_mutex_lock(&schedular->mutex);

    // Find out if the collection_name already exists in the download queue
    list = schedular->queue;
    while (list) {
        collection = (NNTPCollection *) list->data;

        if (!strcmp(collection->collection_name, collection_name)) {
            // Collection found!
            GList *list2;
            gboolean restart_flag = FALSE;

            // Do we have a subject?
            if (!subject) {
                // Restart the whole collection
                restart_flag = TRUE;
            }

            list2 = collection->files;
            while (list2) {
                NNTPFile *file = list2->data;

                if (restart_flag || !strcmp(file->subject, subject)) {
                    // Subject found!
                    GList *list3;
                    gboolean file_is_downloading = FALSE;
                    int filesize_remaining_active = 0;

                    list3 = file->parts;
                    while (list3) {
                        NNTPPart *part = list3->data;

                        if (part->now_downloading) {
                            list3 = g_list_next(list3);
                            file_is_downloading = TRUE;
                            filesize_remaining_active += part->size;
                            continue;
                        }

                        part->downloaded = FALSE;
                        part->now_downloading = FALSE;

                        memset(part->servers_already_tried, 0, sizeof(part->servers_already_tried));

                        list3 = g_list_next(list3);
                    }

                    file->file_is_downloading = file_is_downloading;
                    file->file_is_downloaded = FALSE;
                    file->now_decoding = FALSE;
                    file->num_parts_downloaded = 0;
                    file->num_parts_failed = 0;

                    collection->total_size_remaining += file->file_size - file->file_size_remaining - filesize_remaining_active;
                    file->file_size_remaining = file->file_size - filesize_remaining_active;

                    imported_funcs_task_state_changed(collection_name, file->subject, file->status, TASK_STATE_WAITING_FOR_DOWNLOAD);
                    file->status = TASK_STATE_WAITING_FOR_DOWNLOAD;

                    if (!restart_flag) {
                        g_static_mutex_unlock(&schedular->mutex);

                        return TRUE;
                    }
                }

                list2 = g_list_next(list2);
            }

            if (restart_flag) {
                g_static_mutex_unlock(&schedular->mutex);
                return TRUE;
            }
        }

        list = g_list_next(list);
    }

    g_static_mutex_unlock(&schedular->mutex);

    if (errmsg) {
        *errmsg = g_strdup_printf(_("Subject '%s' was not found in collection '%s'"), subject, collection_name);
    }

    return FALSE;
}

gboolean
nntpgrab_plugin_schedular_save_queue(char **errmsg)
{
    return schedular_plugin_save_queue(errmsg);
}


void
nntpgrab_plugin_schedular_foreach_task(GFunc foreach_func, gpointer data)
{
    g_assert(schedular != NULL);

    g_static_mutex_lock(&schedular->mutex);

    g_list_foreach(schedular->queue, foreach_func, data);

    g_static_mutex_unlock(&schedular->mutex);
}

gboolean
nntpgrab_plugin_schedular_move_task(const char *collection_name_src, const char *subject_src, const char *collection_name_dest, int position_dest)
{
    GList *list;
    NNTPCollection *collection_src = NULL;
    NNTPCollection *collection_dest = NULL;
    NNTPFile *file = NULL;
    gboolean found = FALSE;
    int old_position = 0;

    g_assert(schedular != NULL);

    g_return_val_if_fail(collection_name_src != NULL, FALSE);
    g_return_val_if_fail(subject_src != NULL, FALSE);
    g_return_val_if_fail(collection_name_dest != NULL, FALSE);
    g_return_val_if_fail(position_dest >= -1, FALSE);

    // Moving between collections isn't possible yet in version 0.3
    g_return_val_if_fail(!strcmp(collection_name_src, collection_name_dest), FALSE);

    g_static_mutex_lock(&schedular->mutex);

    // Find the subject
    list = schedular->queue;
    while (list && !found) {
        collection_src = (NNTPCollection *) list->data;

        if (!strcmp(collection_src->collection_name, collection_name_src)) {
            GList *list2;

            list2 = collection_src->files;
            while (list2) {
                file = list2->data;

                if (!strcmp(file->subject, subject_src)) {
                    found = TRUE;
                    break;
                }

                file = NULL;
                list2 = g_list_next(list2);
                old_position++;
            }
        }

        list = g_list_next(list);
    }

    if (!found) {
        g_static_mutex_unlock(&schedular->mutex);
        return FALSE;
    }

    // Is the collection_name_src the same as collection_name_dest ?
    if (!strcmp(collection_name_src, collection_name_dest)) {
        collection_dest = collection_src;
    } else {
        // Nope, we need to find the collection_dest
        list = schedular->queue;
        found = FALSE;
        while (list && !found) {
            collection_dest = (NNTPCollection *) list->data;

            if (!strcmp(collection_dest->collection_name, collection_name_src)) {
                found = TRUE;
            }

            list = g_list_next(list);
        }

        if (!found) {
            g_static_mutex_unlock(&schedular->mutex);
            return FALSE;
        }
    }

    // Perform the actual move
    collection_src->files = g_list_remove(collection_src->files, file);
    collection_dest->files = g_list_insert(collection_dest->files, file, position_dest);

#if 0
    int i = 0;

    list = collection_dest->files;
    while (list) {
        NNTPFile *file = (NNTPFile *) list->data;

        g_print("subject %i = %s\n", i, file->subject);
        i++;
        list = g_list_next(list);
    }
#endif

    g_static_mutex_unlock(&schedular->mutex);

    imported_funcs_emit_task_moved(collection_name_src, subject_src, collection_name_dest, old_position, position_dest);

    return TRUE;
}

gboolean
nntpgrab_plugin_schedular_move_collection(const char *collection_name, int new_position)
{
    GList *list;
    NNTPCollection *collection_src = NULL;
    gboolean found = FALSE;
    int orig_position = 0;

    g_assert(schedular != NULL);

    g_static_mutex_lock(&schedular->mutex);

    // Find the subject
    list = schedular->queue;
    while (list) {
        collection_src = (NNTPCollection *) list->data;

        if (!strcmp(collection_src->collection_name, collection_name)) {
            found = TRUE;
            schedular->queue = g_list_remove(schedular->queue, collection_src);
            schedular->queue = g_list_insert(schedular->queue, collection_src, new_position);
            break;
        }

        orig_position++;
        list = g_list_next(list);
    }

#if 0
    int i = 0;

    list = schedular->queue;
    while (list) {
        collection_src = (NNTPCollection *) list->data;

        g_print("collection %i = %s\n", i, collection_src->collection_name);
        i++;
        list = g_list_next(list);
    }
#endif

    g_static_mutex_unlock(&schedular->mutex);

    if (found) {
        imported_funcs_emit_collection_moved(collection_name, orig_position, new_position);
    }

    return found;
}

gboolean
schedular_plugin_save_queue(char **errmsg)
{
    g_assert(schedular != NULL);

    g_static_mutex_lock(&schedular->mutex);

    if (!save_download_queue(schedular->queue, errmsg)) {
        g_static_mutex_unlock(&schedular->mutex);
        return FALSE;
    }

    g_static_mutex_unlock(&schedular->mutex);

    return TRUE;
}

---