42 #include "MarshallerThread.h"
44 #include "InternalErr.h"
51 bool MarshallerThread::print_time =
false;
58 static double time_diff_to_hundredths(
struct timeval *stop,
struct timeval *start)
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;
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;
72 double result = stop->tv_sec - start->tv_sec;
73 result += double(stop->tv_usec - start->tv_usec) / 1000000;
88 Locker::Locker(pthread_mutex_t &lock, pthread_cond_t &cond,
int &count) :
91 int status = pthread_mutex_lock(&m_mutex);
93 DBG(cerr <<
"Locking the mutex! (waiting; " << pthread_self() <<
")" << endl);
95 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not lock m_mutex");
97 status = pthread_cond_wait(&cond, &m_mutex);
98 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not wait on m_cond");
100 if (count != 0)
throw InternalErr(__FILE__, __LINE__,
"FAIL: left m_cond wait with non-zero child thread count");
102 DBG(cerr <<
"Locked! (" << pthread_self() <<
")" << endl);
110 DBG(cerr <<
"Unlocking the mutex! (" << pthread_self() <<
")" << endl);
112 int status = pthread_mutex_unlock(&m_mutex);
113 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not unlock m_mutex");
130 ChildLocker::ChildLocker(pthread_mutex_t &lock, pthread_cond_t &cond,
int &count) :
131 m_mutex(lock), m_cond(cond), m_count(count)
133 int status = pthread_mutex_lock(&m_mutex);
135 DBG(cerr <<
"Locking the mutex! (simple; " << pthread_self() <<
")" << endl);
137 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not lock m_mutex");
139 DBG(cerr <<
"Locked! (" << pthread_self() <<
")" << endl);
142 ChildLocker::~ChildLocker()
144 DBG(cerr <<
"Unlocking the mutex! (" << pthread_self() <<
")" << endl);
147 int status = pthread_cond_signal(&m_cond);
149 throw InternalErr(__FILE__, __LINE__,
"Could not signal main thread from ChildLocker!");
151 status = pthread_mutex_unlock(&m_mutex);
152 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not unlock m_mutex");
155 MarshallerThread::MarshallerThread() :
156 d_thread(0), d_child_thread_count(0)
158 if (pthread_attr_init(&d_thread_attr) != 0)
throw Error(internal_error,
"Failed to initialize pthread attributes.");
159 if (pthread_attr_setdetachstate(&d_thread_attr, PTHREAD_CREATE_DETACHED ) != 0)
160 throw Error(internal_error,
"Failed to complete pthread attribute initialization.");
162 if (pthread_mutex_init(&d_out_mutex, 0) != 0)
throw Error(internal_error,
"Failed to initialize mutex.");
163 if (pthread_cond_init(&d_out_cond, 0) != 0)
throw Error(internal_error,
"Failed to initialize cond.");
166 MarshallerThread::~MarshallerThread()
168 int status = pthread_mutex_lock(&d_out_mutex);
169 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not lock m_mutex");
170 while (d_child_thread_count != 0) {
171 status = pthread_cond_wait(&d_out_cond, &d_out_mutex);
172 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not wait on m_cond");
174 if (d_child_thread_count != 0)
175 throw InternalErr(__FILE__, __LINE__,
"FAIL: left m_cond wait with non-zero child thread count");
177 status = pthread_mutex_unlock(&d_out_mutex);
178 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not unlock m_mutex");
180 pthread_mutex_destroy(&d_out_mutex);
181 pthread_cond_destroy(&d_out_cond);
183 pthread_attr_destroy(&d_thread_attr);
195 write_args *args =
new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, out, byte_buf,
197 int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
198 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not start child thread");
206 write_args *args =
new write_args(d_out_mutex, d_out_cond, d_child_thread_count, d_thread_error, fd, byte_buf,
208 int status = pthread_create(&d_thread, &d_thread_attr, thread, args);
209 if (status != 0)
throw InternalErr(__FILE__, __LINE__,
"Could not start child thread");
224 write_args *args =
reinterpret_cast<write_args *
>(arg);
226 ChildLocker lock(args->d_mutex, args->d_cond, args->d_count);
230 if (print_time && gettimeofday(&tp_s, 0) != 0) cerr <<
"could not read time" << endl;
236 if (args->d_out_file != -1) {
237 int bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
238 if (bytes_written != args->d_num)
242 args->d_out.write(args->d_buf, args->d_num);
243 if (args->d_out.fail()) {
245 oss <<
"Could not write data: " << __FILE__ <<
":" << __LINE__;
246 args->d_error = oss.str();
251 delete [] args->d_buf;
257 if (gettimeofday(&tp_e, 0) != 0) cerr <<
"could not read time" << endl;
259 cerr <<
"time for child thread write: " << time_diff_to_hundredths(&tp_e, &tp_s) << endl;
281 write_args *args =
reinterpret_cast<write_args *
>(arg);
283 ChildLocker lock(args->d_mutex, args->d_cond, args->d_count);
285 if (args->d_out_file != -1) {
286 int bytes_written = write(args->d_out_file, args->d_buf, args->d_num);
287 if (bytes_written != args->d_num)
return (
void*) -1;
290 args->d_out.write(args->d_buf + 4, args->d_num);
291 if (args->d_out.fail()) {
293 oss <<
"Could not write data: " << __FILE__ <<
":" << __LINE__;
294 args->d_error = oss.str();
299 delete [] args->d_buf;