Commit 0aa77c40 authored by gsamain's avatar gsamain Committed by Xavier Thompson

Acthon unit test

parent 6ce8f80d
# mode: run
# tag: cpp, cpp11, pthread
# cython: experimental_cpp_class_def=True, language_level=2
from libcpp.deque cimport deque
ctypedef deque[ActhonMessageInterface] message_queue_t
cdef extern from "<semaphore.h>" nogil:
ctypedef int sem_t
int sem_init(sem_t *sem, int pshared, unsigned int value)
int sem_wait(sem_t *sem)
int sem_post(sem_t *sem)
int sem_destroy(sem_t* sem)
cdef cypclass BasicQueue(ActhonQueueInterface) checklock:
message_queue_t* _queue
__init__(self):
self._queue = new message_queue_t()
__dealloc__(self):
del self._queue
bint is_empty(self) const:
return self._queue.empty()
void push(self, ActhonMessageInterface message):
self._queue.push_back(message)
if message._sync_method is not NULL:
message._sync_method.insertActivity(message)
Cy_INCREF(message)
bint activate(self):
cdef bint one_message_processed
if self._queue.empty():
return False
# Note here that according to Cython refcount conventions,
# the front() method should have returned a new ref.
# This is obviously not the case, so if we do nothing
# we will, at the end of this function, loose a ref on the pointed object
# (as we will decref the thing pointed by next_message).
next_message = self._queue.front()
self._queue.pop_front()
one_message_processed = next_message.activate()
if one_message_processed:
if next_message._sync_method is not NULL:
with wlocked next_message._sync_method:
next_message._sync_method.removeActivity(next_message)
else:
self._queue.push_back(next_message)
# Don't forget to incref to avoid premature deallocation
Cy_INCREF(next_message)
return one_message_processed
cdef cypclass NoneResult(ActhonResultInterface) checklock:
void pushVoidStarResult(self, void* result):
pass
void pushIntResult(self, int result):
pass
void* getVoidStarResult(self) const:
return NULL
int getIntResult(self) const:
return 0
cdef cypclass WaitResult(ActhonResultInterface) checklock:
union result_t:
int int_val
void* ptr
result_t result
sem_t semaphore
__init__(self):
self.result.ptr = NULL
sem_init(&self.semaphore, 0, 0)
__dealloc__(self):
sem_destroy(&self.semaphore)
@staticmethod
ActhonResultInterface construct():
return WaitResult()
void pushVoidStarResult(self, void* result):
self.result.ptr = result
sem_post(&self.semaphore)
void pushIntResult(self, int result):
self.result.int_val = result
sem_post(&self.semaphore)
result_t _getRawResult(self) const:
# We must ensure a result exists, but we can let others access it immediately
# The cast here is a way of const-casting (we're modifying the semaphore in a const method)
sem_wait(<sem_t*> &self.semaphore)
sem_post(<sem_t*> &self.semaphore)
return self.result
void* getVoidStarResult(self) const:
res = self._getRawResult()
return res.ptr
int getIntResult(self) const:
res = self._getRawResult()
return res.int_val
cdef cypclass ActivityCounterSync(ActhonSyncInterface) checklock:
int count
ActivityCounterSync previous_sync
__init__(self, ActivityCounterSync prev = <ActivityCounterSync> NULL):
self.count = 0
self.previous_sync = prev
void insertActivity(self, ActhonMessageInterface msg):
self.count += 1
void removeActivity(self, ActhonMessageInterface msg):
self.count -= 1
bint isCompleted(self) const:
return self.count == 0
bint isActivable(self) const:
cdef bint res = True
if self.previous_sync is not NULL:
with rlocked self.previous_sync:
res = self.previous_sync.isCompleted()
return res
cdef cypclass A checklock activable:
int a
__init__(self):
self.a = 0
self._active_result_class = WaitResult.construct
self._active_queue_class = BasicQueue()
int getter(self) const:
return self.a
void setter(self, int a):
self.a = a
def test_acthon_chain(n):
"""
>>> test_acthon_chain(42)
42
"""
cdef ActhonResultInterface res
cdef ActhonQueueInterface queue
sync1 = ActivityCounterSync()
with wlocked sync1:
after_sync1 = ActivityCounterSync(sync1)
obj = A()
with wlocked obj:
obj_actor = obj.__activate__()
with wlocked obj_actor, wlocked sync1, wlocked after_sync1:
# Pushing things in the queue
obj_actor.setter(sync1, n)
res = obj_actor.getter(after_sync1)
# Processing the queue
with rlocked obj:
queue = obj._active_queue_class
with wlocked queue:
while not queue.is_empty():
queue.activate()
print <int> res
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment