pdxx
/
libfastcommon
mirror of https://github.com/happyfish100/libfastcommon.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fast_timer.h restore to NO lock version

pull/37/head
YuQing 2020-11-26 17:26:43 +08:00
parent 8718818dc0
commit 1189c97d8e
4 changed files with 59 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
src/fast_timer.c
View File

@ -23,33 +23,15 @@
#include "pthread_func.h"
#include "fast_timer.h"
static int fast_timer_init_locks(FastTimer *timer)
int fast_timer_init(FastTimer *timer, const int slot_count,
const int64_t current_time)
{
int result;
FastTimerSlot *slot;
FastTimerSlot *end;
end = timer->slots + timer->slot_count;
for (slot=timer->slots; slot<end; slot++) {
if ((result=init_pthread_lock(&slot->lock)) != 0) {
return result;
}
}
return 0;
}
int fast_timer_init_ex(FastTimer *timer, const int slot_count,
const int64_t current_time, const bool need_lock)
{
int result;
int bytes;
if (slot_count <= 0 || current_time <= 0) {
return EINVAL;
}
timer->need_lock = need_lock;
timer->slot_count = slot_count;
timer->base_time = current_time; //base time for slot 0
timer->current_time = current_time;
@ -60,84 +42,17 @@ int fast_timer_init_ex(FastTimer *timer, const int slot_count,
}
memset(timer->slots, 0, bytes);
if (need_lock) {
if ((result=fast_timer_init_locks(timer)) != 0) {
return result;
}
}
return 0;
}
void fast_timer_destroy(FastTimer *timer)
{
if (timer->slots != NULL) {
if (timer->need_lock) {
FastTimerSlot *slot;
FastTimerSlot *end;
end = timer->slots + timer->slot_count;
for (slot=timer->slots; slot<end; slot++) {
pthread_mutex_destroy(&slot->lock);
}
}
free(timer->slots);
timer->slots = NULL;
}
}
#define TIMER_CHECK_LOCK(timer, slot) \
do { \
if (timer->need_lock) { \
PTHREAD_MUTEX_LOCK(&(slot)->lock); \
} \
} while (0)
#define TIMER_CHECK_UNLOCK(timer, slot) \
do { \
if (timer->need_lock) { \
PTHREAD_MUTEX_UNLOCK(&(slot)->lock); \
} \
} while (0)
#define TIMER_CHECK_LOCK_AND_SET_SLOT(timer, slot, entry) \
do { \
if (timer->need_lock) { \
PTHREAD_MUTEX_LOCK(&(slot)->lock); \
entry->slot_index = slot - timer->slots; \
} \
} while (0)
#define TIMER_CHECK_LOCK_BY_ENTRY(timer, entry) \
do { \
if (timer->need_lock && entry->slot_index >= 0) { \
PTHREAD_MUTEX_LOCK(&(timer->slots + entry->slot_index)->lock); \
} \
} while (0)
#define TIMER_CHECK_UNLOCK_AND_REMOVE_BY_ENTRY(timer, entry) \
do { \
if (timer->need_lock && entry->slot_index >= 0) { \
PTHREAD_MUTEX_UNLOCK(&(timer->slots + entry->slot_index)->lock); \
entry->slot_index = -1; \
} \
} while (0)
#define TIMER_CHECK_LOCK_BY_SINDEX(timer, slot_index) \
do { \
if (timer->need_lock) { \
PTHREAD_MUTEX_LOCK(&(timer->slots + slot_index)->lock); \
} \
} while (0)
#define TIMER_CHECK_UNLOCK_BY_SINDEX(timer, slot_index) \
do { \
if (timer->need_lock) { \
PTHREAD_MUTEX_UNLOCK(&(timer->slots + slot_index)->lock); \
} \
} while (0)
#define TIMER_GET_SLOT_INDEX(timer, expires) \
(((expires) - timer->base_time) % timer->slot_count)
@ -147,7 +62,6 @@ void fast_timer_destroy(FastTimer *timer)
static inline void add_entry(FastTimer *timer, FastTimerSlot *slot,
FastTimerEntry *entry, const int64_t expires, const bool set_expires)
{
TIMER_CHECK_LOCK_AND_SET_SLOT(timer, slot, entry);
if (set_expires) {
entry->expires = expires;
}
@ -158,7 +72,6 @@ static inline void add_entry(FastTimer *timer, FastTimerSlot *slot,
entry->prev = &slot->head;
slot->head.next = entry;
entry->rehash = false;
TIMER_CHECK_UNLOCK(timer, slot);
}
void fast_timer_add_ex(FastTimer *timer, FastTimerEntry *entry,
@ -179,58 +92,39 @@ void fast_timer_add_ex(FastTimer *timer, FastTimerEntry *entry,
add_entry(timer, slot, entry, new_expires, new_set_expires);
}
void fast_timer_modify(FastTimer *timer, FastTimerEntry *entry,
int fast_timer_modify(FastTimer *timer, FastTimerEntry *entry,
const int64_t new_expires)
{
int slot_index;
int result;
if (new_expires > entry->expires) {
if (timer->need_lock && entry->slot_index >= 0) {
slot_index = entry->slot_index;
if (slot_index < 0) {
slot_index = TIMER_GET_SLOT_INDEX(timer, entry->expires);
}
} else {
slot_index = TIMER_GET_SLOT_INDEX(timer, entry->expires);
}
TIMER_CHECK_LOCK_BY_SINDEX(timer, slot_index);
entry->rehash = TIMER_GET_SLOT_INDEX(timer, new_expires) != slot_index;
entry->rehash = TIMER_GET_SLOT_INDEX(timer, new_expires) !=
TIMER_GET_SLOT_INDEX(timer, entry->expires);
entry->expires = new_expires; //lazy move
TIMER_CHECK_UNLOCK_BY_SINDEX(timer, slot_index);
} else if (new_expires < entry->expires) {
fast_timer_remove(timer, entry);
fast_timer_add_ex(timer, entry, new_expires, true);
if ((result=fast_timer_remove(timer, entry)) == 0) {
fast_timer_add_ex(timer, entry, new_expires, true);
}
return result;
}
}
static inline void remove_entry(FastTimerEntry *entry)
{
if (entry->next != NULL) {
entry->next->prev = entry->prev;
entry->prev->next = entry->next;
entry->next = NULL;
}
else {
entry->prev->next = NULL;
}
entry->prev = NULL;
return 0;
}
int fast_timer_remove(FastTimer *timer, FastTimerEntry *entry)
{
int result;
TIMER_CHECK_LOCK_BY_ENTRY(timer, entry);
if (entry->prev == NULL) {
result = ENOENT; //already removed
} else {
remove_entry(entry);
result = 0;
return ENOENT; //already removed
}
TIMER_CHECK_UNLOCK_AND_REMOVE_BY_ENTRY(timer, entry);
return result;
if (entry->next != NULL) {
entry->next->prev = entry->prev;
entry->prev->next = entry->next;
entry->next = NULL;
} else {
entry->prev->next = NULL;
}
entry->prev = NULL;
return 0;
}
FastTimerSlot *fast_timer_slot_get(FastTimer *timer, const int64_t current_time)
@ -265,7 +159,6 @@ int fast_timer_timeouts_get(FastTimer *timer, const int64_t current_time,
count = 0;
while (timer->current_time < current_time) {
slot = TIMER_GET_SLOT_POINTER(timer, timer->current_time++);
TIMER_CHECK_LOCK(timer, slot);
entry = slot->head.next;
while (entry != NULL) {
if (entry->expires >= current_time) { //not expired
@ -284,9 +177,10 @@ int fast_timer_timeouts_get(FastTimer *timer, const int64_t current_time,
new_slot = TIMER_GET_SLOT_POINTER(timer, last->expires);
if (new_slot != slot) { //check to avoid deadlock
remove_entry(last);
add_entry(timer, new_slot, last,
last->expires, false);
if (fast_timer_remove(timer, last) == 0) {
add_entry(timer, new_slot, last,
last->expires, false);
}
} else {
last->rehash = false;
}
@ -311,7 +205,6 @@ int fast_timer_timeouts_get(FastTimer *timer, const int64_t current_time,
tail = last;
first = NULL;
}
TIMER_CHECK_UNLOCK(timer, slot);
}
if (count > 0) {

11
src/fast_timer.h
View File

@ -31,11 +31,9 @@ typedef struct fast_timer_entry {
typedef struct fast_timer_slot {
struct fast_timer_entry head;
pthread_mutex_t lock;
} FastTimerSlot;
typedef struct fast_timer {
bool need_lock;
int slot_count; //time wheel slot count
int64_t base_time; //base time for slot 0
int64_t current_time;
@ -46,20 +44,17 @@ typedef struct fast_timer {
extern "C" {
#endif
#define fast_timer_init(timer, slot_count, current_time) \
fast_timer_init_ex(timer, slot_count, current_time, false)
#define fast_timer_add(timer, entry) \
fast_timer_add_ex(timer, entry, (entry)->expires, false)
int fast_timer_init_ex(FastTimer *timer, const int slot_count,
const int64_t current_time, const bool need_lock);
int fast_timer_init(FastTimer *timer, const int slot_count,
const int64_t current_time);
void fast_timer_destroy(FastTimer *timer);
void fast_timer_add_ex(FastTimer *timer, FastTimerEntry *entry,
const int64_t expires, const bool set_expires);
int fast_timer_remove(FastTimer *timer, FastTimerEntry *entry);
void fast_timer_modify(FastTimer *timer, FastTimerEntry *entry,
int fast_timer_modify(FastTimer *timer, FastTimerEntry *entry,
const int64_t new_expires);
FastTimerSlot *fast_timer_slot_get(FastTimer *timer, const int64_t current_time);

63
src/locked_timer.c
View File

@ -85,23 +85,24 @@ void locked_timer_destroy(LockedTimer *timer)
pthread_mutex_t *lock;
pthread_mutex_t *lend;
if (timer->slots != NULL) {
send = timer->slots + timer->slot_count;
for (slot=timer->slots; slot<send; slot++) {
pthread_mutex_destroy(&slot->lock);
}
lend = timer->entry_shares.locks + timer->entry_shares.count;
for (lock=timer->entry_shares.locks; lock<lend; lock++) {
pthread_mutex_destroy(lock);
}
free(timer->entry_shares.locks);
timer->entry_shares.locks = NULL;
timer->entry_shares.count = 0;
free(timer->slots);
timer->slots = NULL;
if (timer->slots == NULL) {
return;
}
send = timer->slots + timer->slot_count;
for (slot=timer->slots; slot<send; slot++) {
pthread_mutex_destroy(&slot->lock);
}
lend = timer->entry_shares.locks + timer->entry_shares.count;
for (lock=timer->entry_shares.locks; lock<lend; lock++) {
pthread_mutex_destroy(lock);
}
free(timer->entry_shares.locks);
timer->entry_shares.locks = NULL;
timer->entry_shares.count = 0;
free(timer->slots);
timer->slots = NULL;
}
#define TIMER_GET_SLOT_INDEX(timer, expires) \
@ -110,10 +111,10 @@ void locked_timer_destroy(LockedTimer *timer)
#define TIMER_GET_SLOT_POINTER(timer, expires) \
(timer->slots + TIMER_GET_SLOT_INDEX(timer, expires))
#define LOCKED_TIMER_ENTRY_LOCK(timer, entry) \
#define TIMER_ENTRY_LOCK(timer, entry) \
PTHREAD_MUTEX_LOCK(timer->entry_shares.locks + entry->lock_index)
#define LOCKED_TIMER_ENTRY_UNLOCK(timer, entry) \
#define TIMER_ENTRY_UNLOCK(timer, entry) \
PTHREAD_MUTEX_UNLOCK(timer->entry_shares.locks + entry->lock_index)
static inline void add_entry(LockedTimer *timer, LockedTimerSlot *slot,
@ -125,16 +126,15 @@ static inline void add_entry(LockedTimer *timer, LockedTimerSlot *slot,
timer->entry_shares.count;
}
LOCKED_TIMER_ENTRY_LOCK(timer, entry);
TIMER_ENTRY_LOCK(timer, entry);
entry->status = FAST_TIMER_STATUS_NORMAL;
entry->slot_index = slot - timer->slots;
LOCKED_TIMER_ENTRY_UNLOCK(timer, entry);
TIMER_ENTRY_UNLOCK(timer, entry);
PTHREAD_MUTEX_LOCK(&slot->lock);
if (set_expires) {
entry->expires = expires;
}
fc_list_add_tail(&entry->dlink, &slot->head);
entry->rehash = false;
PTHREAD_MUTEX_UNLOCK(&slot->lock);
@ -150,7 +150,7 @@ static inline int check_set_entry_status(LockedTimer *timer,
int result;
while (1) {
LOCKED_TIMER_ENTRY_LOCK(timer, entry);
TIMER_ENTRY_LOCK(timer, entry);
switch (entry->status) {
case FAST_TIMER_STATUS_CLEARED:
result = ECANCELED;
@ -169,7 +169,7 @@ static inline int check_set_entry_status(LockedTimer *timer,
break;
}
*slot_index = entry->slot_index;
LOCKED_TIMER_ENTRY_UNLOCK(timer, entry);
TIMER_ENTRY_UNLOCK(timer, entry);
if (result == EAGAIN) {
fc_sleep_ms(1);
@ -243,15 +243,6 @@ int locked_timer_remove_ex(LockedTimer *timer, LockedTimerEntry *entry,
return 0;
}
LockedTimerSlot *locked_timer_slot_get(LockedTimer *timer, const int64_t current_time)
{
if (timer->current_time >= current_time) {
return NULL;
}
return TIMER_GET_SLOT_POINTER(timer, timer->current_time++);
}
int locked_timer_timeouts_get(LockedTimer *timer, const int64_t current_time,
LockedTimerEntry *head)
{
@ -278,14 +269,14 @@ int locked_timer_timeouts_get(LockedTimer *timer, const int64_t current_time,
if (entry->rehash) {
new_slot = TIMER_GET_SLOT_POINTER(timer, entry->expires);
if (new_slot != slot) { //check to avoid deadlock
LOCKED_TIMER_ENTRY_LOCK(timer, entry);
TIMER_ENTRY_LOCK(timer, entry);
if (entry->status == FAST_TIMER_STATUS_NORMAL) {
entry->status = FAST_TIMER_STATUS_MOVING;
is_valid = true;
} else {
is_valid = false;
}
LOCKED_TIMER_ENTRY_UNLOCK(timer, entry);
TIMER_ENTRY_UNLOCK(timer, entry);
if (is_valid) {
fc_list_del_init(&entry->dlink);
@ -297,14 +288,14 @@ int locked_timer_timeouts_get(LockedTimer *timer, const int64_t current_time,
}
}
} else { //expired
LOCKED_TIMER_ENTRY_LOCK(timer, entry);
TIMER_ENTRY_LOCK(timer, entry);
if (entry->status == FAST_TIMER_STATUS_NORMAL) {
entry->status = FAST_TIMER_STATUS_TIMEOUT;
is_valid = true;
} else {
is_valid = false;
}
LOCKED_TIMER_ENTRY_UNLOCK(timer, entry);
TIMER_ENTRY_UNLOCK(timer, entry);
if (is_valid) {
fc_list_del_init(&entry->dlink);

6
src/locked_timer.h
View File

@ -27,10 +27,9 @@
#define FAST_TIMER_STATUS_TIMEOUT 3
#define FAST_TIMER_STATUS_CLEARED 4
struct locked_timer_slot;
typedef struct locked_timer_entry {
int64_t expires;
struct fc_list_head dlink;
struct fc_list_head dlink; //for timer slot
struct locked_timer_entry *next; //for timeout chain
uint32_t slot_index; //for slot lock
uint16_t lock_index; //for entry lock
@ -50,7 +49,7 @@ typedef struct locked_timer_shared_locks {
typedef struct locked_timer {
int slot_count; //time wheel slot count
LockedTimerSharedLocks entry_shares;
LockedTimerSharedLocks entry_shares; //shared locks for entry
int64_t base_time; //base time for slot 0
int64_t current_time;
LockedTimerSlot *slots;
@ -80,7 +79,6 @@ int locked_timer_remove_ex(LockedTimer *timer, LockedTimerEntry *entry,
int locked_timer_modify(LockedTimer *timer, LockedTimerEntry *entry,
const int64_t new_expires);
LockedTimerSlot *locked_timer_slot_get(LockedTimer *timer, const int64_t current_time);
int locked_timer_timeouts_get(LockedTimer *timer, const int64_t current_time,
LockedTimerEntry *head);