path: root/src/main/cpp/threadcxx.cpp

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <log4cxx/logstring.h>
#include <log4cxx/helpers/thread.h>
#include <log4cxx/helpers/exception.h>
#include <apr_thread_proc.h>
#include <apr_atomic.h>
#include <log4cxx/helpers/pool.h>
#include <log4cxx/helpers/threadlocal.h>
#include <log4cxx/helpers/synchronized.h>
#include <apr_thread_cond.h>

using namespace log4cxx::helpers;
using namespace log4cxx;

#if APR_HAS_THREADS
namespace {
                        /**
                         *   This class is used to encapsulate the parameters to
                         *   Thread::run when they are passed to Thread::launcher.
                         *
                         */
                        class LaunchPackage {
                        public:
                            /**
                             *  Placement new to create LaunchPackage in specified pool.
                             *  LaunchPackage needs to be dynamically allocated since
                             *  since a stack allocated instance may go out of scope
                             *  before thread is launched.
                             */
                            static void* operator new(size_t sz, Pool& p) {
    				return p.palloc(sz);
			    }
                            /**
                            *  operator delete would be called if exception during construction.
                            */
                            static void operator delete(void*, Pool& p) {
                            }
                            /**
                             *  Create new instance.
                             */
			    LaunchPackage(Thread* t, Runnable r, void* d) : thread(t), runnable(r), data(d) {
                            }
                            /**
                             * Gets thread parameter.
                             * @return thread.
                             */
                            Thread* getThread() const {
				  return thread;
			    }

                            /**
                             *  Gets runnable parameter.
                             *  @return runnable.
                             */
                            Runnable getRunnable() const {
                                  return runnable;
                            }
                            /**
                             *  gets data parameter.
                             *  @return thread.
                             */
                            void* getData() const {
                                  return data;
                            }
                        private:
                            LaunchPackage(const LaunchPackage&);
                            LaunchPackage& operator=(const LaunchPackage&);
                            Thread* thread;
                            Runnable runnable; 
                            void* data;
                        };
                        
                        /**
                         *  This object atomically sets the specified memory location
                         *  to non-zero on construction and to zero on destruction.  
                         *  Used to maintain Thread.alive.
                         */
                        class LaunchStatus {
                        public:
                            /*
                             *  Construct new instance.
                             *  @param p address of memory to set to non-zero on construction, zero on destruction.
                             */
			    LaunchStatus(volatile unsigned int* p) : alive(p) {
    				apr_atomic_set32(alive, 0xFFFFFFFF);
			    }
                            /**
                             *  Destructor.
                             */
                            ~LaunchStatus() {
				 apr_atomic_set32(alive, 0);
                             }

                        private:
                            LaunchStatus(const LaunchStatus&);
                            LaunchStatus& operator=(const LaunchStatus&);
                            volatile unsigned int* alive;
                        };

                        /**
                         *   Get a key to the thread local storage used to hold the reference to
                         *   the corresponding Thread object.
                         */                        
			ThreadLocal& getThreadLocal() {
     				static ThreadLocal tls;
     				return tls;
			}

}

void* LOG4CXX_THREAD_FUNC ThreadLaunch::launcher(apr_thread_t* thread, void* data) {
	LaunchPackage* package = (LaunchPackage*) data;
	ThreadLocal& tls = getThreadLocal();
	tls.set(package->getThread());
    {
      (package->getRunnable())(thread, package->getData());
      package->getThread()->ending();
    }
    apr_thread_exit(thread, 0); // this function never returns !
    return 0;
}
#endif

Thread::Thread() : thread(NULL), alive(0), interruptedStatus(0), 
    interruptedMutex(NULL), interruptedCondition(NULL) {
}

Thread::~Thread() {
    join();
}



void Thread::run(Runnable start, void* data) {
#if APR_HAS_THREADS
        // Try to join first if previous instance did exit
        if ( isActive() && !isAlive() ) {
             join();
        }
        // now we're ready to create the thread again
        //
        //    if attempting a second run method on the same Thread object
        //         throw an exception
        //
        if (thread != NULL) {
            throw IllegalStateException();
        }
        apr_threadattr_t* attrs;
        apr_status_t stat = apr_threadattr_create(&attrs, p.getAPRPool());
        if (stat != APR_SUCCESS) {
                throw ThreadException(stat);
        }
        
       stat = apr_thread_cond_create(&interruptedCondition, p.getAPRPool());
       if (stat != APR_SUCCESS) {
            throw ThreadException(stat);
       }
       stat = apr_thread_mutex_create(&interruptedMutex, APR_THREAD_MUTEX_NESTED, 
                    p.getAPRPool());
       if (stat != APR_SUCCESS) {
            throw ThreadException(stat);
       }
        
        //   create LaunchPackage on the thread's memory pool
        LaunchPackage* package = new(p) LaunchPackage(this, start, data);
        stat = apr_thread_create(&thread, attrs,
            ThreadLaunch::launcher, package, p.getAPRPool());
        if (stat != APR_SUCCESS) {
                throw ThreadException(stat);
        }
        // we need to set alive here already, since we use isAlive() to check
        // if run() has been called in a thread-safe way.
        apr_atomic_set32(&alive, 0xFFFFFFFF);
#else
        throw ThreadException(LOG4CXX_STR("APR_HAS_THREADS is not true"));
#endif
}

void Thread::join() {
#if APR_HAS_THREADS
    if (thread != NULL) {
        apr_status_t startStat;
        apr_status_t stat = apr_thread_join(&startStat, thread);
        thread = NULL;
        if (stat != APR_SUCCESS) {
            throw ThreadException(stat);
        }
    }
#endif
}

void Thread::currentThreadInterrupt() {
#if APR_HAS_THREADS
   void* tls = getThreadLocal().get();
   if (tls != 0) {
       ((Thread*) tls)->interrupt();
   }
#endif
}

void Thread::interrupt() {
    apr_atomic_set32(&interruptedStatus, 0xFFFFFFFF);
#if APR_HAS_THREADS
    if (interruptedMutex != NULL) {
        synchronized sync(interruptedMutex);
        apr_status_t stat = apr_thread_cond_signal(interruptedCondition);
        if (stat != APR_SUCCESS) throw ThreadException(stat);
    }
#endif    
}

bool Thread::interrupted() {
#if APR_HAS_THREADS
   void* tls = getThreadLocal().get();
   if (tls != 0) {
       return apr_atomic_xchg32(&(((Thread*) tls)->interruptedStatus), 0) != 0;
   }
#endif
   return false;
}

bool Thread::isCurrentThread() const {
#if APR_HAS_THREADS
    const void* tls = getThreadLocal().get();
    return (tls == this);
#else
    return true;
#endif
}

bool Thread::isAlive() {
    return apr_atomic_read32(&alive) != 0;
}

void Thread::ending() {
    apr_atomic_set32(&alive, 0);
}


void Thread::sleep(int duration) {
#if APR_HAS_THREADS
    if(interrupted()) {
         throw InterruptedException();
    }
    if (duration > 0) {
        Thread* pThis = (Thread*) getThreadLocal().get();
        if (pThis == NULL) {
            apr_sleep(duration*1000);
        } else {
            synchronized sync(pThis->interruptedMutex);
            apr_status_t stat = apr_thread_cond_timedwait(pThis->interruptedCondition, 
                pThis->interruptedMutex, duration*1000);
            if (stat != APR_SUCCESS && !APR_STATUS_IS_TIMEUP(stat)) {
                throw ThreadException(stat);
            }
            if (interrupted()) {
                throw InterruptedException();
            }
        }
    }
#else    
    if (duration > 0) {
        apr_sleep(duration*1000);
    }
#endif    
}