Line data Source code
1 : /*
2 : * AsynchronousTools.cc
3 : *
4 : * Created on: Nov 1, 2010
5 : * Author: jjacobs
6 : */
7 :
8 : #include <assert.h>
9 : #include <cstdarg>
10 : #include <cstring>
11 : #include <errno.h>
12 : #include <fstream>
13 : #include <queue>
14 : #include <semaphore.h>
15 : #include <fcntl.h>
16 :
17 : #include <time.h>
18 : #include <casacore/casa/aips.h>
19 : #if defined(AIPS_LINUX)
20 : #if ! defined(_GNU_SOURCE)
21 : #define _GNU_SOURCE /* or _BSD_SOURCE or _SVID_SOURCE */
22 : #endif
23 : #include <unistd.h>
24 : #include <sys/syscall.h> /* For SYS_xxx definitions */
25 : #endif
26 : #include <sys/time.h>
27 :
28 : #include <casacore/casa/Exceptions/Error.h>
29 : #include <casacore/casa/Logging/LogIO.h>
30 :
31 : #include <condition_variable>
32 : #include <mutex>
33 : #include <thread>
34 :
35 : #include "AsynchronousTools.h"
36 : #include <stdcasa/UtilJ.h>
37 :
38 : using namespace std;
39 : using namespace casacore;
40 : using namespace casa::utilj;
41 :
42 : namespace casa {
43 :
44 : namespace async {
45 :
46 : class ConditionImpl {
47 :
48 : friend class Condition;
49 :
50 : private:
51 :
52 0 : ConditionImpl () : condition_p () {}
53 :
54 : std::condition_variable condition_p;
55 : };
56 :
57 : class MutexImpl {
58 :
59 : friend class Mutex;
60 : friend class Condition;
61 :
62 : private:
63 :
64 6 : MutexImpl () : mutex_p () {}
65 6 : ~MutexImpl () {}
66 :
67 : std::thread::id lockingThreadId_p;
68 : std::mutex mutex_p;
69 : };
70 :
71 : class SemaphoreImpl {
72 :
73 : friend class Semaphore;
74 :
75 : private:
76 :
77 0 : SemaphoreImpl () : semaphore_p (NULL) {}
78 :
79 : sem_t * semaphore_p; // [use]
80 : };
81 :
82 : struct timespec
83 0 : convertMsDeltaToTimespec (Int milliseconds)
84 : {
85 : // Get the time with a possible accuracy to microseconds and
86 : // then convert it into the timeout's data structure used by
87 : // pthreads
88 :
89 : struct timeval tVal;
90 0 : gettimeofday (& tVal, NULL);
91 :
92 : struct timespec t;
93 0 : t.tv_sec = tVal.tv_sec;
94 0 : t.tv_nsec = tVal.tv_usec * 1000;
95 :
96 : // Add the wait time in milliseconds to this structure
97 : // taking care to handle carry out and avoid overflow
98 :
99 0 : t.tv_sec += milliseconds / 1000;
100 : // extract seconds and add them in
101 :
102 0 : t.tv_nsec += (milliseconds % 1000) * 1000000L;
103 : // extract ms, convert to ns and add in
104 :
105 0 : t.tv_sec += t.tv_nsec / 1000000000L;
106 : // carry out of the ns field into seconds
107 :
108 0 : t.tv_nsec %= 1000000000L;
109 : // remove any seconds carried out of ns field
110 :
111 0 : return t;
112 : }
113 :
114 :
115 0 : Condition::Condition ()
116 : {
117 0 : impl_p = new ConditionImpl ();
118 0 : }
119 :
120 0 : Condition::~Condition ()
121 : {
122 0 : delete impl_p;
123 0 : }
124 :
125 : void
126 0 : Condition::broadcast ()
127 : {
128 0 : notify_all ();
129 0 : }
130 :
131 : void
132 0 : Condition::notify_all ()
133 : {
134 0 : impl_p->condition_p.notify_all ();
135 0 : }
136 :
137 : void
138 0 : Condition::notify_one ()
139 : {
140 0 : impl_p->condition_p.notify_one ();
141 0 : }
142 :
143 : void
144 0 : Condition::signal ()
145 : {
146 0 : notify_one ();
147 0 : }
148 :
149 : void
150 0 : Condition::wait (UniqueLock & uniqueLock)
151 : {
152 0 : impl_p->condition_p.wait (uniqueLock.uniqueLock_p);
153 0 : }
154 : /*
155 : Bool
156 : Condition::wait (Mutex & mutex, int milliseconds)
157 : {
158 : Assert (milliseconds >= 0); // weird if it's negative
159 :
160 : struct timespec t = convertMsDeltaToTimespec (milliseconds);
161 : int code = pthread_cond_timedwait (impl_p->condition_p, mutex.getRep(), & t);
162 :
163 : bool gotWait = true;
164 : if (code == ETIMEDOUT){
165 : gotWait = false;
166 : }
167 : else{
168 : ThrowIfError (code, String::format ("Condition::wait (%d)", milliseconds));
169 : }
170 :
171 : return gotWait;
172 : }
173 : */
174 :
175 0 : LockGuard::LockGuard (Mutex & mutex)
176 : {
177 0 : mutex_p = & mutex;
178 0 : mutex_p->lock ();
179 0 : }
180 :
181 0 : LockGuard::LockGuard (Mutex * mutex)
182 : {
183 0 : Assert (mutex != NULL);
184 :
185 0 : mutex_p = mutex;
186 0 : mutex_p->lock ();
187 0 : }
188 :
189 0 : LockGuard::~LockGuard ()
190 : {
191 0 : mutex_p->unlock ();
192 0 : }
193 :
194 0 : LockGuardInverse::LockGuardInverse (Mutex & mutex)
195 : {
196 0 : mutex_p = & mutex;
197 0 : mutex_p->unlock ();
198 0 : }
199 :
200 0 : LockGuardInverse::LockGuardInverse (Mutex * mutex)
201 : {
202 0 : Assert (mutex != NULL);
203 :
204 0 : mutex_p = mutex;
205 0 : mutex_p->unlock ();
206 0 : }
207 :
208 0 : LockGuardInverse::LockGuardInverse (LockGuard & lg)
209 : {
210 0 : mutex_p = lg.mutex_p;
211 0 : mutex_p->unlock();
212 0 : }
213 :
214 :
215 0 : LockGuardInverse::~LockGuardInverse ()
216 : {
217 0 : mutex_p->lock ();
218 0 : }
219 :
220 :
221 :
222 : Logger* Logger::singleton_p = NULL;
223 :
224 0 : Logger::Logger ()
225 0 : : loggingStarted_p (false),
226 0 : nameMutex_p (new Mutex ())
227 0 : {}
228 :
229 0 : Logger::~Logger ()
230 : {
231 0 : if (loggingStarted_p){
232 0 : delete get();
233 : }
234 0 : }
235 :
236 : std::once_flag loggerOnceFlag;
237 :
238 : Logger*
239 0 : Logger::get()
240 : {
241 0 : std::call_once (loggerOnceFlag, initialize);
242 :
243 0 : return singleton_p;
244 : }
245 :
246 : void
247 0 : Logger::initialize ()
248 : {
249 0 : singleton_p = new Logger ();
250 0 : }
251 :
252 : void
253 0 : Logger::log (const char * format, ...)
254 : {
255 : va_list vaList;
256 0 : va_start (vaList, format);
257 :
258 : char buffer[4096];
259 :
260 : // Create the text to be logged
261 :
262 0 : vsnprintf (buffer, sizeof (buffer), format, vaList);
263 :
264 : // Grab the timestamp and pid (for IDing threads)
265 :
266 0 : String threadNameText;
267 :
268 : {
269 0 : MutexLocker ml (* nameMutex_p);
270 :
271 0 : pthread_t tid = pthread_self();
272 0 : ThreadNames::iterator threadName = threadNames_p.find (tid);
273 0 : if (threadName != threadNames_p.end ()){
274 0 : threadNameText = String (" [") + (threadName->second) + "] : ";
275 : }
276 : else{
277 0 : threadNameText = String::format (" [0x%08x] : ", tid);
278 : }
279 0 : }
280 :
281 0 : String prefix = utilj::getTimestamp() + threadNameText;
282 :
283 : // Allocate a buffer to put into the queue
284 :
285 0 : string outputText = prefix + buffer;
286 :
287 0 : va_end (vaList);
288 :
289 : // Lock the queue, push on the block of text and increment
290 : // the drain semaphore
291 :
292 0 : loggerThread_p -> log (outputText); // ownership passes to the thread
293 0 : }
294 :
295 : void
296 0 : Logger::registerName (const String & threadName)
297 : {
298 0 : Assert (nameMutex_p != NULL);
299 :
300 0 : MutexLocker ml (* nameMutex_p);
301 :
302 0 : threadNames_p [pthread_self()] = threadName;
303 0 : }
304 :
305 : void
306 0 : Logger::start (const char * filename)
307 : {
308 0 : if (! loggingStarted_p){ // ignore multiple starts
309 :
310 0 : loggerThread_p = new LoggerThread ();
311 :
312 0 : loggerThread_p ->setLogFilename (filename == NULL ? "" : filename);
313 :
314 0 : loggerThread_p ->startThread();
315 :
316 0 : loggingStarted_p = true;
317 : }
318 0 : }
319 :
320 0 : Logger::LoggerThread::LoggerThread ()
321 0 : {}
322 :
323 0 : Logger::LoggerThread::~LoggerThread ()
324 : {
325 0 : terminate();
326 :
327 0 : this->join();
328 :
329 0 : if (deleteStream_p)
330 : {
331 0 : dynamic_cast<ofstream *> (logStream_p)->close();
332 0 : delete logStream_p;
333 : }
334 0 : }
335 :
336 : void
337 0 : Logger::LoggerThread::log (const string & text)
338 : {
339 0 : MutexLocker m (mutex_p);
340 :
341 0 : outputQueue_p.push (text);
342 :
343 0 : loggerChanged_p.notify_all ();
344 0 : }
345 :
346 :
347 : void *
348 0 : Logger::LoggerThread::run ()
349 : {
350 0 : LogIO logIo (LogOrigin ("Logger::LoggerThread"));
351 0 : logIo << "starting execution; tid=" << gettid() << endl << LogIO::POST;
352 :
353 :
354 : try {
355 : // Determine where to write the logging info. If nothing is specified or either "cerr" or
356 : // "stdout" are specified then use standard error. If "cout" or "stdout" are specified then
357 : // use standard out. Otherwise open the specified file and write to that.
358 :
359 0 : if (logFilename_p.empty () || logFilename_p == "cerr" || logFilename_p == "stderr"){
360 0 : logStream_p = & cerr;
361 0 : deleteStream_p = false;
362 : }
363 0 : else if (logFilename_p == "cout" || logFilename_p == "stdout"){
364 0 : logStream_p = & cout;
365 0 : deleteStream_p = false;
366 : }
367 : else{
368 0 : logStream_p = new ofstream (logFilename_p.c_str(), ios::out);
369 0 : deleteStream_p = true;
370 : }
371 :
372 0 : * logStream_p << utilj::getTimestamp() << ": Logging started, tid=" << gettid() << endl;
373 :
374 : // Loop waiting on the drain semaphore. This should be incremented once
375 : // every time users add a block of text to the queue.
376 :
377 : while (true){
378 :
379 0 : string text;
380 :
381 : {
382 : // Pop the front block of output off of the queue
383 : // Keep mutex locked while accessing queue.
384 :
385 0 : UniqueLock uniqueLock (mutex_p);
386 :
387 0 : while (! isTerminationRequested() && outputQueue_p.empty()){
388 0 : loggerChanged_p.wait (uniqueLock);
389 : }
390 :
391 0 : if (isTerminationRequested() && outputQueue_p.empty()){
392 0 : break;
393 : }
394 :
395 0 : text = outputQueue_p.front();
396 :
397 0 : outputQueue_p.pop();
398 0 : }
399 :
400 : // Now output the text and then delete the storage
401 :
402 0 : * logStream_p << text;
403 :
404 0 : logStream_p->flush();
405 0 : }
406 :
407 0 : * logStream_p << "*** Logging terminated" << endl;
408 :
409 0 : logStream_p->flush();
410 :
411 0 : return NULL;
412 : }
413 0 : catch (exception & e){
414 :
415 0 : const char * message = "*** Logging thread caught exception: ";
416 :
417 0 : cerr << message << e.what() << endl;
418 0 : cerr.flush();
419 :
420 0 : if (logStream_p != & cerr){
421 :
422 0 : * logStream_p << message << e.what() << endl;
423 0 : logStream_p->flush();
424 : }
425 :
426 0 : throw;
427 0 : }
428 0 : catch (...){
429 :
430 0 : const char * message = "*** Logging thread caught unknown exception";
431 :
432 0 : cerr << message << endl;
433 0 : cerr.flush();
434 :
435 0 : if (logStream_p != & cerr){
436 0 : * logStream_p << message << endl;
437 0 : logStream_p->flush();
438 : }
439 :
440 0 : throw;
441 0 : }
442 0 : }
443 :
444 : void
445 0 : Logger::LoggerThread::setLogFilename (const String & filename)
446 : {
447 0 : logFilename_p = filename;
448 0 : }
449 :
450 : void
451 0 : Logger::LoggerThread::terminate ()
452 : {
453 0 : Thread::terminate();
454 :
455 0 : loggerChanged_p.notify_all ();
456 0 : }
457 :
458 6 : Mutex::Mutex ()
459 : {
460 6 : impl_p = new MutexImpl ();
461 6 : isLocked_p = false;
462 6 : }
463 :
464 6 : Mutex::~Mutex ()
465 : {
466 6 : delete impl_p;
467 6 : }
468 :
469 : std::mutex &
470 0 : Mutex::getMutex ()
471 : {
472 0 : return impl_p->mutex_p;
473 : }
474 :
475 : //Bool
476 : //Mutex::isLockedByThisThread () const
477 : //{
478 : // // Only for use in debugs or asserts
479 : //
480 : // Bool itIs = isLocked_p && std::this_thread::get_id () == impl_p->lockingThreadId_p;
481 : //
482 : // return itIs;
483 : //}
484 :
485 : void
486 0 : Mutex::lock ()
487 : {
488 0 : impl_p->mutex_p.lock();
489 0 : impl_p->lockingThreadId_p = std::this_thread::get_id ();
490 0 : isLocked_p = true;
491 0 : }
492 :
493 : /*
494 : Bool
495 : Mutex::lock (Int milliseconds)
496 : {
497 :
498 : Assert (milliseconds >= 0); // weird if it's negative
499 :
500 : struct timespec t = convertMsDeltaToTimespec (milliseconds);
501 : int code = pthread_mutex_timedlock (impl_p->mutex_p, & t);
502 :
503 : bool gotLock = true;
504 : if (code == ETIMEDOUT){
505 : gotLock = false;
506 : }
507 : else{
508 : ThrowIfError (code, String::format ("Mutex::lock (%d)", milliseconds));
509 : }
510 :
511 : return gotLock;
512 : }
513 : */
514 :
515 : Bool
516 30 : Mutex::trylock ()
517 : {
518 30 : bool gotLock = impl_p->mutex_p.try_lock ();
519 30 : isLocked_p = gotLock;
520 30 : if (isLocked_p){
521 30 : impl_p->lockingThreadId_p = std::this_thread::get_id ();
522 : }
523 :
524 30 : return gotLock;
525 : }
526 :
527 : void
528 30 : Mutex::unlock ()
529 : {
530 30 : isLocked_p = false;
531 30 : impl_p->mutex_p.unlock ();
532 30 : }
533 :
534 : // jagonzal: Useful when locking is mandatory
535 : void
536 30 : Mutex::acquirelock()
537 : {
538 30 : while (!trylock())
539 : {
540 0 : sched_yield();
541 : }
542 30 : }
543 :
544 0 : MutexLocker::MutexLocker (Mutex & mutex)
545 0 : : mutex_p (& mutex)
546 : {
547 0 : mutex_p->lock();
548 0 : }
549 :
550 0 : MutexLocker::MutexLocker (Mutex * mutex)
551 0 : : mutex_p (mutex)
552 : {
553 0 : Assert (mutex_p != NULL);
554 :
555 0 : mutex_p->lock();
556 0 : }
557 :
558 0 : MutexLocker::~MutexLocker ()
559 : {
560 0 : mutex_p->unlock();
561 0 : }
562 :
563 0 : Semaphore::Semaphore (int initialValue)
564 : {
565 0 : Assert (initialValue >= 0);
566 :
567 0 : impl_p = new SemaphoreImpl ();
568 :
569 : // Since Mac doesn't support unnamed semaphores, try and find a
570 : // unique name for the semaphore. Names will be of the form
571 : // "/Semaphore_xxx"
572 :
573 0 : int code = 0;
574 0 : int i = 0;
575 :
576 : do {
577 :
578 0 : ++ i;
579 :
580 0 : name_p = String::format ("/CasaAsync_%03d", i);
581 0 : impl_p->semaphore_p = sem_open (name_p.c_str(), O_CREAT | O_EXCL, 0700, initialValue);//new sem_t;
582 0 : code = (impl_p->semaphore_p == SEM_FAILED) ? errno : 0;
583 :
584 0 : } while (impl_p->semaphore_p == SEM_FAILED && code == EEXIST);
585 :
586 0 : ThrowIfError (code, "Semaphore::open: name='" + name_p + "'");
587 0 : }
588 :
589 0 : Semaphore::~Semaphore ()
590 : {
591 0 : int code = sem_close (impl_p->semaphore_p);
592 0 : ThrowIfError (code == 0 ? 0 : errno, "Semaphore::close");
593 :
594 0 : code = sem_unlink (name_p.c_str());
595 0 : ThrowIfError (code == 0 ? 0 : errno, "Semaphore::unlink: name='" + name_p + "'");
596 :
597 0 : delete impl_p;
598 0 : }
599 :
600 : Int
601 0 : Semaphore::getValue ()
602 : {
603 : int value;
604 0 : int code = sem_getvalue (impl_p->semaphore_p, & value);
605 0 : ThrowIfError (code == 0 ? 0 : errno, "Semaphore::getValue");
606 :
607 0 : return value;
608 : }
609 :
610 : void
611 0 : Semaphore::post ()
612 : {
613 0 : int code = sem_post (impl_p->semaphore_p);
614 0 : ThrowIfError (code == 0 ? 0 : errno, "Semaphore::post");
615 0 : }
616 :
617 : Bool
618 0 : Semaphore::trywait ()
619 : {
620 0 : int code = sem_trywait (impl_p->semaphore_p);
621 0 : bool gotSemaphore = true;
622 :
623 0 : if (code != 0 && errno == EAGAIN){
624 0 : gotSemaphore = false;
625 : }
626 : else{
627 0 : ThrowIfError (code == 0 ? 0 : errno, "Semaphore::wait");
628 : }
629 :
630 0 : return gotSemaphore;
631 : }
632 :
633 : void
634 0 : Semaphore::wait ()
635 : {
636 0 : int errorCode = 0;
637 : int code;
638 :
639 : do {
640 :
641 0 : code = sem_wait (impl_p->semaphore_p);
642 0 : errorCode = errno;
643 :
644 0 : } while (code != 0 && errorCode == EINTR);
645 :
646 0 : ThrowIfError (code == 0 ? 0 : errorCode, "Semaphore::wait");
647 0 : }
648 :
649 : Bool
650 0 : Semaphore::wait (int milliseconds)
651 : {
652 0 : Assert (milliseconds >= 0); // it's weird if it's negative
653 :
654 : //// struct timespec t = convertMsDeltaToTimespec (milliseconds);
655 0 : int errorCode = 0;
656 : int code;
657 :
658 : do {
659 :
660 0 : code = sem_wait (impl_p->semaphore_p);
661 0 : errorCode = errno;
662 :
663 0 : } while (code != 0 && errorCode == EINTR);
664 :
665 :
666 0 : Bool gotSemaphore = true;
667 :
668 0 : if (code == 0){
669 0 : gotSemaphore = true;
670 0 : } else if (errno == ETIMEDOUT){
671 0 : gotSemaphore = false;
672 : } else {
673 0 : ThrowIfError (errno, String::format ("Mutex::lock (%d)", milliseconds));
674 : }
675 :
676 0 : return gotSemaphore;
677 : }
678 :
679 :
680 1148 : Thread::Thread ()
681 : {
682 1148 : id_p = new pthread_t;
683 1148 : started_p = false;
684 1148 : terminationRequested_p = false;
685 1148 : }
686 :
687 1148 : Thread::~Thread ()
688 : {
689 : // Make sure the thread knows it's time to quit
690 :
691 1148 : terminate ();
692 :
693 1148 : delete id_p;
694 1148 : }
695 :
696 : pthread_t
697 0 : Thread::getId () const
698 : {
699 0 : return * id_p;
700 : }
701 :
702 : pid_t
703 0 : Thread::gettid () const
704 : {
705 0 : pid_t result = 0;
706 : #if defined(AIPS_LINUX)
707 0 : result = syscall (SYS_gettid);
708 : #endif
709 0 : return result;
710 : }
711 :
712 : void *
713 0 : Thread::join ()
714 : {
715 : void * result;
716 0 : int code = pthread_join (* id_p, & result);
717 0 : ThrowIfError (code, "Thread::join");
718 :
719 0 : return result;
720 : }
721 :
722 : bool
723 0 : Thread::isStarted () const
724 : {
725 0 : return started_p;
726 : }
727 :
728 : void
729 0 : Thread::startThread ()
730 : {
731 : // Create the thread, passing a pointer to this object as its
732 : // single argument. Subclass Thread to pass other information
733 : // into the thread function.
734 :
735 0 : int code = pthread_create (id_p, NULL, threadFunction, this);
736 0 : started_p = true;
737 0 : ThrowIfError (code, "Thread::create");
738 0 : }
739 :
740 : void
741 1148 : Thread::terminate ()
742 : {
743 1148 : terminationRequested_p = true;
744 1148 : }
745 :
746 : bool
747 0 : Thread::isTerminationRequested () const
748 : {
749 0 : return terminationRequested_p;
750 : }
751 :
752 : void *
753 0 : Thread::threadFunction (void * arg)
754 : {
755 0 : Thread * thread = reinterpret_cast<Thread *> (arg);
756 :
757 0 : void * result = thread->run ();
758 :
759 0 : return result; // use thread variable to store any results
760 : }
761 :
762 0 : UniqueLock::UniqueLock (Mutex & mutex)
763 0 : : uniqueLock_p (mutex.getMutex())
764 0 : {}
765 :
766 : void
767 0 : UniqueLock::lock ()
768 : {
769 0 : uniqueLock_p.lock ();
770 0 : }
771 :
772 : void
773 0 : UniqueLock::unlock ()
774 : {
775 0 : uniqueLock_p.unlock ();
776 0 : }
777 :
778 :
779 :
780 : } // end namespace Async
781 :
782 : } // end namespace CASA
|
