src/usr/util/threadpool.C


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//  IBM_PROLOG_BEGIN_TAG
//  This is an automatically generated prolog.
//
//  $Source: src/usr/util/threadpool.C $
//
//  IBM CONFIDENTIAL
//
//  COPYRIGHT International Business Machines Corp. 2012
//
//  p1
//
//  Object Code Only (OCO) source materials
//  Licensed Internal Code Source Materials
//  IBM HostBoot Licensed Internal Code
//
//  The source code for this program is not published or other-
//  wise divested of its trade secrets, irrespective of what has
//  been deposited with the U.S. Copyright Office.
//
//  Origin: 30
//
//  IBM_PROLOG_END
#include <util/threadpool.H>
#include <sys/task.h>
#include <sys/misc.h>

void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__init()
{
    // Initialize all member variables.
    iv_worklist.clear();
    mutex_init(&iv_mutex);
    sync_cond_init(&iv_condvar);
    iv_children.clear();
    iv_shutdown = false;
}

void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__insert(void* i_workItem)
{
    mutex_lock(&iv_mutex);

    // Insert item on to the work list and signal any worker thread that may
    // be waiting.
    iv_worklist.push_back(i_workItem);
    sync_cond_signal(&iv_condvar);

    mutex_unlock(&iv_mutex);
}

void* Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__remove(order_fn_t fn)
{
    void* val = NULL;

    mutex_lock(&iv_mutex);
    do
    {
        // Wait until the worklist is empty or told to shutdown.
        while(iv_worklist.empty() && !iv_shutdown)
        {
            sync_cond_wait(&iv_condvar, &iv_mutex);
        }

        // If told to shutdown and no work remains, end thread.
        if (iv_worklist.empty() && iv_shutdown)
        {
            break;
        }

        // Otherwise, obtain next item from worklist, using order function
        // passed to us.
        worklist_itr_t itr = fn(iv_worklist.begin(), iv_worklist.end());

        val = *itr;
        iv_worklist.erase(itr);

    } while(0);

    mutex_unlock(&iv_mutex);

    return val;
}

void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__start(
        Util::__Util_ThreadPool_Impl::ThreadPoolImpl::start_fn_t fn,
        void* instance)
{
    mutex_lock(&iv_mutex);

    size_t thread_count = Util::ThreadPoolManager::getThreadCount();
    while(thread_count--)
    {
        // Create children and add to a queue for later joining (during
        // shutdown)
        iv_children.push_back(task_create(fn, instance));
    }

    mutex_unlock(&iv_mutex);
}

void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__shutdown()
{
    mutex_lock(&iv_mutex);

    // Set shutdown status and signal all children to release from their
    // condition variable.
    iv_shutdown = true;
    sync_cond_broadcast(&iv_condvar);

    // Join on all the children.
    while(!iv_children.empty())
    {
        tid_t child = iv_children.front();
        iv_children.pop_front();

        mutex_unlock(&iv_mutex);
        task_wait_tid(child, NULL, NULL);  // Don't need status.
        mutex_lock(&iv_mutex);
    }

    mutex_unlock(&iv_mutex);
}

// Default thread count of one per HW thread.
size_t Util::ThreadPoolManager::cv_size = cpu_thread_count();