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
124
125
126
127
128
|
/*****************************************************************
* gmerlin - a general purpose multimedia framework and applications
*
* Copyright (c) 2001 - 2010 Members of the Gmerlin project
* gmerlin-general@lists.sourceforge.net
* http://gmerlin.sourceforge.net
*
* 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, see <http://www.gnu.org/licenses/>.
* *****************************************************************/
#include <pthread.h>
#include <stdlib.h>
#include <gavl/gavl.h>
#include <gmerlin/parameter.h>
#include <gmerlin/bggavl.h>
#include <gmerlin/bg_sem.h>
typedef struct
{
pthread_t t;
sem_t run_sem;
sem_t done_sem;
pthread_mutex_t stop_mutex;
int do_stop;
void (*func)(void*, int, int);
void * data;
int start;
int len;
} thread_t;
struct bg_thread_pool_s
{
int num_threads;
thread_t * threads;
};
static void * thread_func(void * data)
{
thread_t * t = data;
int do_stop;
while(1)
{
sem_wait(&t->run_sem);
pthread_mutex_lock(&t->stop_mutex);
do_stop = t->do_stop;
pthread_mutex_unlock(&t->stop_mutex);
if(do_stop)
break;
t->func(t->data, t->start, t->len);
sem_post(&t->done_sem);
}
return NULL;
}
bg_thread_pool_t * bg_thread_pool_create(int num_threads)
{
int i;
bg_thread_pool_t * ret = calloc(1, sizeof(*ret));
ret->num_threads = num_threads;
ret->threads = calloc(num_threads, sizeof(*ret->threads));
for(i = 0; i < ret->num_threads; i++)
{
pthread_mutex_init(&ret->threads[i].stop_mutex, NULL);
sem_init(&ret->threads[i].run_sem, 0, 0);
sem_init(&ret->threads[i].done_sem, 0, 0);
pthread_create(&ret->threads[i].t,
(pthread_attr_t*)0,
thread_func, &ret->threads[i]);
}
return ret;
}
void bg_thread_pool_destroy(bg_thread_pool_t * p)
{
int i;
for(i = 0; i < p->num_threads; i++)
{
pthread_mutex_lock(&p->threads[i].stop_mutex);
p->threads[i].do_stop = 1;
pthread_mutex_unlock(&p->threads[i].stop_mutex);
sem_post(&p->threads[i].run_sem);
pthread_join(p->threads[i].t, NULL);
pthread_mutex_destroy(&p->threads[i].stop_mutex);
sem_destroy(&p->threads[i].run_sem);
sem_destroy(&p->threads[i].done_sem);
}
free(p->threads);
free(p);
}
void bg_thread_pool_run(void (*func)(void*,int start, int len),
void * gavl_data,
int start, int len,
void * client_data, int thread)
{
bg_thread_pool_t * p = client_data;
p->threads[thread].func = func;
p->threads[thread].data = gavl_data;
p->threads[thread].start = start;
p->threads[thread].len = len;
sem_post(&p->threads[thread].run_sem);
}
void bg_thread_pool_stop(void * client_data, int thread)
{
bg_thread_pool_t * p = client_data;
sem_wait(&p->threads[thread].done_sem);
}
|
