libdap Updated for version 3.21.0
libdap4 is an implementation of OPeNDAP's DAP protocol.
MarshallerThread.cc
1// -*- mode: c++; c-basic-offset:4 -*-
2
3// This file is part of libdap, A C++ implementation of the OPeNDAP Data
4// Access Protocol.
5
6// Copyright (c) 2015 OPeNDAP, Inc.
7// Author: James Gallagher <jgallagher@opendap.org>
8//
9// This library is free software; you can redistribute it and/or
10// modify it under the terms of the GNU Lesser General Public
11// License as published by the Free Software Foundation; either
12// version 2.1 of the License, or (at your option) any later version.
13//
14// This library is distributed in the hope that it will be useful,
15// but WITHOUT ANY WARRANTY; without even the implied warranty of
16// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17// Lesser General Public License for more details.
18//
19// You should have received a copy of the GNU Lesser General Public
20// License along with this library; if not, write to the Free Software
21// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
22//
23// You can contact OPeNDAP, Inc. at PO Box 112, Saunderstown, RI. 02874-0112.
24
25/*
26 * MarshallerThread.cc
27 *
28 * Created on: Aug 27, 2015
29 * Author: jimg
30 */
31
32#include "config.h"
33
34#include <fcntl.h>
35#include <pthread.h>
36#include <sys/time.h>
37#include <unistd.h>
38
39#include <ostream>
40#include <sstream>
41
42#include "Error.h"
43#include "InternalErr.h"
44#include "MarshallerThread.h"
45#include "debug.h"
46
47using namespace libdap;
48using namespace std;
49
50#if 0
51bool MarshallerThread::print_time = false;
52
58static double time_diff_to_hundredths(struct timeval *stop, struct timeval *start)
59{
60 /* Perform the carry for the later subtraction by updating y. */
61 if (stop->tv_usec < start->tv_usec) {
62 int nsec = (start->tv_usec - stop->tv_usec) / 1000000 + 1;
63 start->tv_usec -= 1000000 * nsec;
64 start->tv_sec += nsec;
65 }
66 if (stop->tv_usec - start->tv_usec > 1000000) {
67 int nsec = (start->tv_usec - stop->tv_usec) / 1000000;
68 start->tv_usec += 1000000 * nsec;
69 start->tv_sec -= nsec;
70 }
71
72 double result = stop->tv_sec - start->tv_sec;
73 result += double(stop->tv_usec - start->tv_usec) / 1000000;
74 return result;
75}
76#endif
77
87Locker::Locker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count) : m_mutex(lock) {
88 int status = pthread_mutex_lock(&m_mutex);
89
90 DBG(cerr << "Locking the mutex! (waiting; " << pthread_self() << ")" << endl);
91
92 if (status != 0)
93 throw InternalErr(__FILE__, __LINE__, "Could not lock m_mutex");
94 while (count != 0) {
95 status = pthread_cond_wait(&cond, &m_mutex);
96 if (status != 0)
97 throw InternalErr(__FILE__, __LINE__, "Could not wait on m_cond");
98 }
99 if (count != 0)
100 throw InternalErr(__FILE__, __LINE__, "FAIL: left m_cond wait with non-zero child thread count");
101
102 DBG(cerr << "Locked! (" << pthread_self() << ")" << endl);
103}
104
109 DBG(cerr << "Unlocking the mutex! (" << pthread_self() << ")" << endl);
110
111 (void)pthread_mutex_unlock(&m_mutex);
112#if 0
113 int status = pthread_mutex_unlock(&m_mutex);
114 if (status != 0) throw InternalErr(__FILE__, __LINE__, "Could not unlock m_mutex");
115#endif
116}
117
131ChildLocker::ChildLocker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count)
132 : m_mutex(lock), m_cond(cond), m_count(count) {
133 int status = pthread_mutex_lock(&m_mutex);
134
135 DBG(cerr << "Locking the mutex! (simple; " << pthread_self() << ")" << endl);
136
137 if (status != 0)
138 throw InternalErr(__FILE__, __LINE__, "Could not lock m_mutex");
139
140 DBG(cerr << "Locked! (" << pthread_self() << ")" << endl);
141}
142
143ChildLocker::~ChildLocker() {
144 DBG(cerr << "Unlocking the mutex! (" << pthread_self() << ")" << endl);
145
146 m_count = 0;
147
148 (void)pthread_cond_signal(&m_cond);
149 (void)pthread_mutex_unlock(&m_mutex);
150
151#if 0
152 int status = pthread_cond_signal(&m_cond);
153 if (status != 0)
154 throw InternalErr(__FILE__, __LINE__, "Could not signal main thread from ChildLocker!");
155
156 status = pthread_mutex_unlock(&m_mutex);
157 if (status != 0) throw InternalErr(__FILE__, __LINE__, "Could not unlock m_mutex");
158#endif
159}
160
161MarshallerThread::MarshallerThread() : d_thread(0), d_child_thread_count(0) {
162 if (pthread_attr_init(&d_thread_attr) != 0)
163 throw Error(internal_error, "Failed to initialize pthread attributes.");
164 if (pthread_attr_setdetachstate(&d_thread_attr, PTHREAD_CREATE_DETACHED /*PTHREAD_CREATE_JOINABLE*/) != 0)
165 throw Error(internal_error, "Failed to complete pthread attribute initialization.");
166
167 if (pthread_mutex_init(&d_out_mutex, 0) != 0)
168 throw Error(internal_error, "Failed to initialize mutex.");
169 if (pthread_cond_init(&d_out_cond, 0) != 0)
170 throw Error(internal_error, "Failed to initialize cond.");
171}
172
173MarshallerThread::~MarshallerThread() {
174 (void)pthread_mutex_lock(&d_out_mutex);
175#if 0
176 int status = pthread_mutex_lock(&d_out_mutex);
177 if (status != 0) throw InternalErr(__FILE__, __LINE__, "Could not lock m_mutex");
178#endif
179 // d_child_thread_count is passed into the thread in a structure (see write_thread())
180 // and is decremented by the ChildLocker dtor when write_thread() exits. jhrg 2/7/19
181 if (d_child_thread_count != 0) {
182 (void)pthread_cond_wait(&d_out_cond, &d_out_mutex);
183 d_child_thread_count = 0;
184#if 0
185 status = pthread_cond_wait(&d_out_cond, &d_out_mutex);
186 if (status != 0) throw InternalErr(__FILE__, __LINE__, "Could not wait on m_cond");
187#endif
188 }
189
190 (void)pthread_mutex_unlock(&d_out_mutex);
191
192#if 0
193 if (d_child_thread_count != 0)
194 throw InternalErr(__FILE__, __LINE__, "FAIL: left m_cond wait with non-zero child thread count");
195
196 status = pthread_mutex_unlock(&d_out_mutex);
197 if (status != 0) throw InternalErr(__FILE__, __LINE__, "Could not unlock m_mutex");
198#endif
199
200 pthread_mutex_destroy(&d_out_mutex);
201 pthread_cond_destroy(&d_out_cond);
202
203 pthread_attr_destroy(&d_thread_attr);
204}
205
206// not a static method
212void MarshallerThread::start_thread(void *(*thread)(void *arg), ostream &out, char *byte_buf, unsigned int bytes) {
213 write_args *args =
214 new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, out, byte_buf, bytes);
215 int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
216 if (status != 0)
217 throw InternalErr(__FILE__, __LINE__, "Could not start child thread");
218}
219
223void MarshallerThread::start_thread(void *(*thread)(void *arg), int fd, char *byte_buf, unsigned int bytes) {
224 write_args *args =
225 new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, fd, byte_buf, bytes);
226 int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
227 if (status != 0)
228 throw InternalErr(__FILE__, __LINE__, "Could not start child thread");
229}
230
241 write_args *args = reinterpret_cast<write_args *>(arg);
242
243 ChildLocker lock(args->d_mutex, args->d_cond, args->d_count); // RAII; will unlock on exit
244
245#if 0
246 struct timeval tp_s;
247 if (print_time && gettimeofday(&tp_s, 0) != 0) cerr << "could not read time" << endl;
248#endif
249
250 // force an error
251 // return (void*)-1;
252
253 if (args->d_out_file != -1) {
254 int bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
255 if (bytes_written != args->d_num)
256 return (void *)-1;
257 } else {
258 args->d_out.write(args->d_buf, args->d_num);
259 if (args->d_out.fail()) {
260 ostringstream oss;
261 oss << "Could not write data: " << __FILE__ << ":" << __LINE__;
262 args->d_error = oss.str();
263 return (void *)-1;
264 }
265 }
266
267 delete[] args->d_buf;
268 delete args;
269
270#if 0
271 struct timeval tp_e;
272 if (print_time) {
273 if (gettimeofday(&tp_e, 0) != 0) cerr << "could not read time" << endl;
274
275 cerr << "time for child thread write: " << time_diff_to_hundredths(&tp_e, &tp_s) << endl;
276 }
277#endif
278
279 return 0;
280}
281
295 write_args *args = reinterpret_cast<write_args *>(arg);
296
297 ChildLocker lock(args->d_mutex, args->d_cond, args->d_count); // RAII; will unlock on exit
298
299 if (args->d_out_file != -1) {
300 int bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
301 if (bytes_written != args->d_num)
302 return (void *)-1;
303 } else {
304 args->d_out.write(args->d_buf + 4, args->d_num);
305 if (args->d_out.fail()) {
306 ostringstream oss;
307 oss << "Could not write data: " << __FILE__ << ":" << __LINE__;
308 args->d_error = oss.str();
309 return (void *)-1;
310 }
311 }
312
313 delete[] args->d_buf;
314 delete args;
315
316 return 0;
317}
ChildLocker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count)
A class for error processing.
Definition Error.h:92
A class for software fault reporting.
Definition InternalErr.h:61
Locker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count)
static void * write_thread(void *arg)
static void * write_thread_part(void *arg)
void start_thread(void *(*thread)(void *arg), std::ostream &out, char *byte_buf, unsigned int bytes_written)
top level DAP object to house generic methods
Definition AISConnect.cc:30