46ad30c008223ac208766118ac23524c28a37f47
[pazpar2-moved-to-github.git] / src / sel_thread.c
1 /* $Id: sel_thread.c,v 1.2 2007-04-20 10:15:19 adam Exp $
2    Copyright (c) 2006-2007, Index Data.
3
4 This file is part of Pazpar2.
5
6 Pazpar2 is free software; you can redistribute it and/or modify it under
7 the terms of the GNU General Public License as published by the Free
8 Software Foundation; either version 2, or (at your option) any later
9 version.
10
11 Pazpar2 is distributed in the hope that it will be useful, but WITHOUT ANY
12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
14 for more details.
15
16 You should have received a copy of the GNU General Public License
17 along with Pazpar2; see the file LICENSE.  If not, write to the
18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
19 02111-1307, USA.
20  */
21
22 #if HAVE_CONFIG_H
23 #include "cconfig.h"
24 #endif
25
26 #include "sel_thread.h"
27 #include <yaz/log.h>
28 #include <yaz/nmem.h>
29 #include <unistd.h>
30 #include <stdlib.h>
31 #include <pthread.h>
32
33 struct work_item {
34     void *data;
35     struct work_item *next;
36 };
37
38 struct sel_thread {
39     int fd[2];
40     NMEM nmem;
41     pthread_t thread_id;
42     pthread_mutex_t mutex;
43     pthread_cond_t input_data;
44     int stop_flag;
45     struct work_item *input_queue;
46     struct work_item *output_queue;
47     struct work_item *free_queue;
48     void (*work_handler)(void *work_data);;
49 };
50
51 static void *sel_thread_handler(void *vp)
52 {
53     sel_thread_t p = (sel_thread_t) vp;
54
55     while(1)
56     {
57         struct work_item **work_p, *work_this;
58         /* wait for some work */
59         pthread_mutex_lock(&p->mutex);
60         while (!p->stop_flag && !p->input_queue)
61             pthread_cond_wait(&p->input_data, &p->mutex);
62         /* see if we were waken up because we're shutting down */
63         if (p->stop_flag)
64             break;
65         /* got something. Take the last one out of input_queue */
66         work_p = &p->input_queue;
67         while ((*work_p)->next)
68             work_p = &(*work_p)->next;
69         work_this = *work_p;
70         *work_p = 0;
71         pthread_mutex_unlock(&p->mutex);
72
73         /* work on this item */
74         p->work_handler(work_this->data);
75
76         /* put it back into output queue */
77         pthread_mutex_lock(&p->mutex);
78         work_this->next = p->output_queue;
79         p->output_queue = work_this;
80         pthread_mutex_unlock(&p->mutex);
81
82         /* wake up select/poll with a single byte */
83         write(p->fd[1], "", 1);
84     }        
85     pthread_mutex_unlock(&p->mutex);
86     return 0;
87 }
88
89 sel_thread_t sel_thread_create(void (*work_handler)(void *work_data),
90                                int *read_fd)
91 {
92     NMEM nmem = nmem_create();
93     sel_thread_t p = nmem_malloc(nmem, sizeof(*p));
94
95     p->nmem = nmem;
96     if (pipe(p->fd))
97     {
98         nmem_destroy(nmem);
99         return 0;
100     }
101     *read_fd = p->fd[0];
102     p->input_queue = 0;
103     p->output_queue = 0;
104     p->free_queue = 0;
105     p->work_handler = work_handler;
106
107     p->stop_flag = 0;
108     pthread_mutex_init(&p->mutex, 0);
109     pthread_cond_init(&p->input_data, 0);
110     pthread_create (&p->thread_id, 0, sel_thread_handler, p);
111     return p;
112 }
113
114 void sel_thread_destroy(sel_thread_t p)
115 {
116     pthread_mutex_lock(&p->mutex);
117
118     p->stop_flag = 1;
119     pthread_cond_broadcast(&p->input_data);
120     pthread_mutex_unlock(&p->mutex);
121     
122     pthread_join(p->thread_id, 0);
123
124     close(p->fd[0]);
125     close(p->fd[1]);
126     pthread_cond_destroy(&p->input_data);
127     pthread_mutex_destroy(&p->mutex);
128     nmem_destroy(p->nmem);
129 }
130
131 void sel_thread_add(sel_thread_t p, void *data)
132 {
133     struct work_item *work_p;
134
135     pthread_mutex_lock(&p->mutex);
136     work_p = p->free_queue;
137     if (!work_p)
138         work_p = nmem_malloc(p->nmem, sizeof(*work_p));
139
140     work_p->data = data;
141     work_p->next = p->input_queue;
142     p->input_queue = work_p;
143     pthread_mutex_unlock(&p->mutex);
144 }
145
146 void *sel_thread_result(sel_thread_t p)
147 {
148     struct work_item **work_p, *work_this;
149     void *data;
150     char read_buf[1];
151
152     /* got something. Take the last one out of output_queue */
153     work_p = &p->output_queue;
154     if (!*work_p)
155         return 0;
156
157     read(p->fd[0], read_buf, 1);
158
159     while ((*work_p)->next)
160         work_p = &(*work_p)->next;
161     work_this = *work_p;
162     *work_p = 0;
163
164     /* put freed item in free list */
165     work_this->next = p->free_queue;
166     p->free_queue = work_this;
167
168     data = work_this->data;
169     pthread_mutex_unlock(&p->mutex);
170
171     return data;
172 }
173
174 /*
175  * Local variables:
176  * c-basic-offset: 4
177  * indent-tabs-mode: nil
178  * End:
179  * vim: shiftwidth=4 tabstop=8 expandtab
180  */