Statistics
| Branch: | Revision:

ffmpeg / libavcodec / pthread.c @ da9c9637

History | View | Annotate | Download (4.53 KB)

1
/*
2
 * Copyright (c) 2004 Roman Shaposhnik.
3
 * 
4
 * Many thanks to Steven M. Schultz for providing clever ideas and
5
 * to Michael Niedermayer <michaelni@gmx.at> for writing initial
6
 * implementation.
7
 *
8
 * This library is free software; you can redistribute it and/or
9
 * modify it under the terms of the GNU Lesser General Public
10
 * License as published by the Free Software Foundation; either
11
 * version 2 of the License, or (at your option) any later version.
12
 *
13
 * This library is distributed in the hope that it will be useful,
14
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
16
 * Lesser General Public License for more details.
17
 *
18
 * You should have received a copy of the GNU Lesser General Public
19
 * License along with this library; if not, write to the Free Software
20
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
21
 *
22
 */
23
#include <pthread.h>
24

    
25
#include "avcodec.h"
26
#include "common.h"
27

    
28
typedef int (action_t)(AVCodecContext *c, void *arg);
29

    
30
typedef struct ThreadContext {
31
    pthread_t *workers;
32
    action_t *func;
33
    void **args;
34
    int *rets;
35
    int rets_count;
36
    int job_count;
37
    
38
    pthread_cond_t last_job_cond;
39
    pthread_cond_t current_job_cond;
40
    pthread_mutex_t current_job_lock;
41
    int current_job;
42
    int done;
43
} ThreadContext;
44

    
45
static void* worker(void *v) 
46
{
47
    AVCodecContext *avctx = v;
48
    ThreadContext *c = avctx->thread_opaque;
49
    int our_job = c->job_count;
50
    int thread_count = avctx->thread_count;
51
    int self_id;
52

    
53
    pthread_mutex_lock(&c->current_job_lock);
54
    self_id = c->current_job++;
55
    for (;;){ 
56
        while (our_job >= c->job_count) {
57
            if (c->current_job == thread_count + c->job_count)
58
                pthread_cond_signal(&c->last_job_cond);
59
            
60
            pthread_cond_wait(&c->current_job_cond, &c->current_job_lock);
61
            our_job = self_id;
62
            
63
            if (c->done) {
64
                pthread_mutex_unlock(&c->current_job_lock);
65
                return NULL;
66
            }
67
        }
68
        pthread_mutex_unlock(&c->current_job_lock);
69
        
70
        c->rets[our_job%c->rets_count] = c->func(avctx, c->args[our_job]);
71
        
72
        pthread_mutex_lock(&c->current_job_lock);
73
        our_job = c->current_job++;
74
    }
75
}
76

    
77
static always_inline void avcodec_thread_park_workers(ThreadContext *c, int thread_count)
78
{
79
    pthread_cond_wait(&c->last_job_cond, &c->current_job_lock);
80
    pthread_mutex_unlock(&c->current_job_lock);
81
}
82

    
83
void avcodec_thread_free(AVCodecContext *avctx) 
84
{
85
    ThreadContext *c = avctx->thread_opaque;
86
    int i;
87
    
88
    pthread_mutex_lock(&c->current_job_lock);
89
    c->done = 1;
90
    pthread_cond_broadcast(&c->current_job_cond);
91
    pthread_mutex_unlock(&c->current_job_lock);
92

    
93
    for (i=0; i<avctx->thread_count; i++)
94
         pthread_join(c->workers[i], NULL);
95

    
96
    pthread_mutex_destroy(&c->current_job_lock);
97
    pthread_cond_destroy(&c->current_job_cond);
98
    pthread_cond_destroy(&c->last_job_cond);
99
    av_free(c->workers);
100
    av_free(c);
101
}
102

    
103
int avcodec_thread_execute(AVCodecContext *avctx, action_t* func, void **arg, int *ret, int job_count) 
104
{
105
    ThreadContext *c= avctx->thread_opaque;
106
    int dummy_ret;
107
   
108
    if (job_count <= 0)
109
        return 0;
110
    
111
    pthread_mutex_lock(&c->current_job_lock);
112

    
113
    c->current_job = avctx->thread_count;
114
    c->job_count = job_count;
115
    c->args = arg;
116
    c->func = func;
117
    if (ret) {
118
        c->rets = ret;
119
        c->rets_count = job_count;
120
    } else { 
121
        c->rets = &dummy_ret;
122
        c->rets_count = 1;
123
    }
124
    pthread_cond_broadcast(&c->current_job_cond);
125

    
126
    avcodec_thread_park_workers(c, avctx->thread_count);
127
    
128
    return 0;
129
}
130

    
131
int avcodec_thread_init(AVCodecContext *avctx, int thread_count) 
132
{
133
    int i;
134
    ThreadContext *c;
135

    
136
    c = av_mallocz(sizeof(ThreadContext));
137
    if (!c)
138
        return -1;
139
        
140
    c->workers = av_mallocz(sizeof(pthread_t)*thread_count);
141
    if (!c->workers) {
142
        av_free(c);
143
        return -1;
144
    }
145

    
146
    avctx->thread_opaque = c;
147
    avctx->thread_count = thread_count;
148
    c->current_job = 0;
149
    c->job_count = 0;
150
    c->done = 0;
151
    pthread_cond_init(&c->current_job_cond, NULL); 
152
    pthread_cond_init(&c->last_job_cond, NULL);
153
    pthread_mutex_init(&c->current_job_lock, NULL);
154
    pthread_mutex_lock(&c->current_job_lock);
155
    for (i=0; i<thread_count; i++) { 
156
        if(pthread_create(&c->workers[i], NULL, worker, avctx)) {
157
           avctx->thread_count = i;
158
           pthread_mutex_unlock(&c->current_job_lock);
159
           avcodec_thread_free(avctx);
160
           return -1;
161
        }
162
    }
163
    
164
    avcodec_thread_park_workers(c, thread_count);
165
    
166
    avctx->execute = avcodec_thread_execute;
167
    return 0;
168
}