MarshallerThread.cc
// -*- mode: c++; c-basic-offset:4 -*-
// This file is part of libdap, A C++ implementation of the OPeNDAP Data
// Access Protocol.
// Copyright (c) 2015 OPeNDAP, Inc.
// Author: James Gallagher <jgallagher@opendap.org>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This library 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
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
//
// You can contact OPeNDAP, Inc. at PO Box 112, Saunderstown, RI. 02874-0112.
/*
* MarshallerThread.cc
*
* Created on: Aug 27, 2015
* Author: jimg
*/
#include "config.h"
#include <fcntl.h>
#include <pthread.h>
#include <sys/time.h>
#include <unistd.h>
#include <ostream>
#include <sstream>
#include "Error.h"
#include "InternalErr.h"
#include "MarshallerThread.h"
#include "debug.h"
using namespace libdap;
using namespace std;
// Set TIMING to 1 to enable timing output
#define TIMING 0
#if TIMING
bool MarshallerThread::print_time = false;
static double time_diff_to_hundredths(struct timeval *stop, struct timeval *start) {
/* Perform the carry for the later subtraction by updating y. */
if (stop->tv_usec < start->tv_usec) {
int nsec = (start->tv_usec - stop->tv_usec) / 1000000 + 1;
start->tv_usec -= 1000000 * nsec;
start->tv_sec += nsec;
}
if (stop->tv_usec - start->tv_usec > 1000000) {
int nsec = (start->tv_usec - stop->tv_usec) / 1000000;
start->tv_usec += 1000000 * nsec;
start->tv_sec -= nsec;
}
double result = stop->tv_sec - start->tv_sec;
result += double(stop->tv_usec - start->tv_usec) / 1000000;
return result;
}
#endif
Locker::Locker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count) : m_mutex(lock) {
int status = pthread_mutex_lock(&m_mutex);
DBG(cerr << "Locking the mutex! (waiting; " << pthread_self() << ")" << endl);
if (status != 0)
throw InternalErr(__FILE__, __LINE__, "Could not lock m_mutex");
while (count != 0) {
status = pthread_cond_wait(&cond, &m_mutex);
if (status != 0)
throw InternalErr(__FILE__, __LINE__, "Could not wait on m_cond");
}
DBG(cerr << "Locked! (" << pthread_self() << ")" << endl);
}
Locker::~Locker() {
DBG(cerr << "Unlocking the mutex! (" << pthread_self() << ")" << endl);
(void)pthread_mutex_unlock(&m_mutex);
}
ChildLocker::ChildLocker(pthread_mutex_t &lock, pthread_cond_t &cond, int &count)
: m_mutex(lock), m_cond(cond), m_count(count) {
int status = pthread_mutex_lock(&m_mutex);
DBG(cerr << "Locking the mutex! (simple; " << pthread_self() << ")" << endl);
if (status != 0)
throw InternalErr(__FILE__, __LINE__, "Could not lock m_mutex");
DBG(cerr << "Locked! (" << pthread_self() << ")" << endl);
}
ChildLocker::~ChildLocker() {
DBG(cerr << "Unlocking the mutex! (" << pthread_self() << ")" << endl);
m_count = 0;
(void)pthread_cond_signal(&m_cond);
(void)pthread_mutex_unlock(&m_mutex);
}
MarshallerThread::MarshallerThread() {
if (pthread_attr_init(&d_thread_attr) != 0)
throw Error(internal_error, "Failed to initialize pthread attributes.");
if (pthread_attr_setdetachstate(&d_thread_attr, PTHREAD_CREATE_DETACHED /*PTHREAD_CREATE_JOINABLE*/) != 0)
throw Error(internal_error, "Failed to complete pthread attribute initialization.");
if (pthread_mutex_init(&d_out_mutex, nullptr) != 0)
throw Error(internal_error, "Failed to initialize mutex.");
if (pthread_cond_init(&d_out_cond, nullptr) != 0)
throw Error(internal_error, "Failed to initialize cond.");
}
MarshallerThread::~MarshallerThread() {
(void)pthread_mutex_lock(&d_out_mutex);
// d_child_thread_count is passed into the thread in a structure (see write_thread())
// and is decremented by the ChildLocker dtor when write_thread() exits. jhrg 2/7/19
if (d_child_thread_count != 0) {
(void)pthread_cond_wait(&d_out_cond, &d_out_mutex);
d_child_thread_count = 0;
}
(void)pthread_mutex_unlock(&d_out_mutex);
pthread_mutex_destroy(&d_out_mutex);
pthread_cond_destroy(&d_out_cond);
pthread_attr_destroy(&d_thread_attr);
}
// not a static method
void MarshallerThread::start_thread(void *(*thread)(void *arg), ostream &out, char *byte_buf, std::streamsize bytes) {
auto *args = new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, out, byte_buf, bytes);
int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
if (status != 0)
throw InternalErr(__FILE__, __LINE__, "Could not start child thread");
}
void MarshallerThread::start_thread(void *(*thread)(void *arg), int fd, char *byte_buf, std::streamsize bytes) {
auto *args = new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, fd, byte_buf, bytes);
int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
if (status != 0)
throw InternalErr(__FILE__, __LINE__, "Could not start child thread");
}
void *MarshallerThread::write_thread(void *arg) {
auto *args = reinterpret_cast<write_args *>(arg);
ChildLocker lock(args->d_mutex, args->d_cond, args->d_count); // RAII; will unlock on exit
#if TIMING
struct timeval tp_s;
if (print_time && gettimeofday(&tp_s, 0) != 0)
cerr << "could not read time" << endl;
#endif
// force an error
// return (void*)-1;
if (args->d_out_file != -1) {
auto bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
if (bytes_written != args->d_num)
return (void *)-1;
} else {
args->d_out.write(args->d_buf, args->d_num);
if (args->d_out.fail()) {
ostringstream oss;
oss << "Could not write data: " << __FILE__ << ":" << __LINE__;
args->d_error = oss.str();
return (void *)-1;
}
}
delete[] args->d_buf;
delete args;
#if TIMING
struct timeval tp_e;
if (print_time) {
if (gettimeofday(&tp_e, 0) != 0)
cerr << "could not read time" << endl;
cerr << "time for child thread write: " << time_diff_to_hundredths(&tp_e, &tp_s) << endl;
}
#endif
return nullptr;
}
void *MarshallerThread::write_thread_part(void *arg) {
auto *args = reinterpret_cast<write_args *>(arg);
ChildLocker lock(args->d_mutex, args->d_cond, args->d_count); // RAII; will unlock on exit
if (args->d_out_file != -1) {
auto bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
if (bytes_written != args->d_num)
return (void *)-1;
} else {
args->d_out.write(args->d_buf + 4, args->d_num);
if (args->d_out.fail()) {
ostringstream oss;
oss << "Could not write data: " << __FILE__ << ":" << __LINE__;
args->d_error = oss.str();
return (void *)-1;
}
}
delete[] args->d_buf;
delete args;
return nullptr;
}
libdap::ChildLocker
Definition MarshallerThread.h:74
libdap::Error
A class for error processing.
Definition Error.h:92
libdap::InternalErr
A class for software fault reporting.
Definition InternalErr.h:61
libdap::Locker::~Locker
virtual ~Locker()
Definition MarshallerThread.cc:108
libdap::MarshallerThread::write_thread
static void * write_thread(void *arg)
Definition MarshallerThread.cc:210
libdap::MarshallerThread::start_thread
void start_thread(void *(*thread)(void *arg), std::ostream &out, char *byte_buf, std::streamsize bytes_written)
Definition MarshallerThread.cc:184
libdap::MarshallerThread::write_thread_part
static void * write_thread_part(void *arg)
Definition MarshallerThread.cc:266
libdap
top level DAP object to house generic methods
Definition AlarmHandler.h:35