@ -223,18 +223,22 @@ void queue_entry_free(struct ll_entry* entry) {
// Внутренняя функция добавления в хеш-таблицу
static void add_to_hash ( struct ll_queue * q , struct ll_entry * entry ) {
if ( ! q | | q - > hash_size = = 0 | | ! entry ) return ;
size_t slot = entry - > id % q - > hash_size ;
if ( ! q | | ! entry ) return ;
if ( q - > hash_size = = 0 | | entry - > index_size = = 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_LL_QUEUE , " [%s] Empty hash_size=%d or index_size=%d " , q - > name , q - > hash_size , entry - > index_size ) ;
return ;
}
uint32_t slot = entry - > index_hash % q - > hash_size ;
entry - > hash_next = q - > hash_table [ slot ] ;
q - > hash_table [ slot ] = entry ;
}
// Внутренняя функция удаления из хеш-таблицы
static void remove_from_hash ( struct ll_queue * q , struct ll_entry * entry ) {
if ( ! q | | q - > hash_size = = 0 | | ! entry ) return ;
size _tslot = entry - > id % q - > hash_size ;
uint32 _tslot = entry - > in dex_hash % q - > hash_size ; // теперь используем index_hash
struct ll_entry * * ptr = & q - > hash_table [ slot ] ;
while ( * ptr ) {
if ( * ptr = = entry ) {
@ -246,6 +250,20 @@ static void remove_from_hash(struct ll_queue* q, struct ll_entry* entry) {
}
}
static uint32_t make_hash ( const void * data , uint16_t len ) { // алгоритм FNV-1a
if ( len = = 0 | | data = = NULL ) return 0 ;
const uint8_t * p = ( const uint8_t * ) data ;
uint32_t hash = 0x811C9DC5u ; // FNV-1a offset basis (32-bit)
for ( uint16_t i = 0 ; i < len ; + + i ) {
hash ^ = ( uint32_t ) p [ i ] ; // XOR с байтом
hash * = 0x01000193u ; // FNV-1a prime
}
return hash ;
}
// Проверить и запустить ожидающие коллбэки
static void check_waiters ( struct ll_queue * q ) {
if ( ! q ) return ;
@ -267,118 +285,185 @@ static void check_waiters(struct ll_queue* q) {
}
}
int queue_data_put ( struct ll_queue * q , struct ll_entry * entry , uint32_t id ) {
int queue_data_put ( struct ll_queue * q , struct ll_entry * entry ) {
if ( ! q | | ! entry ) return - 1 ;
# ifdef QUEUE_THREAD_CHECK
queue_check_thread ( q ) ;
# endif
entry - > index_offset = 0 ;
entry - > index_size = 0 ;
entry - > index_hash = 0 ;
if ( q - > hash_size > 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_LL_QUEUE , " [%s] Put no-hash data to hash queue " , q - > name ) ;
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
if ( q - > size_limit > = 0 & & q - > count > = q - > size_limit ) {
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
// Добавляем в конец
entry - > next = NULL ;
entry - > prev = q - > tail ;
if ( q - > tail ) q - > tail - > next = entry ; else q - > head = entry ;
q - > tail = entry ;
q - > count + + ;
entry - > int_len = entry - > len ;
q - > total_bytes + = entry - > int_len ;
// НЕ вызываем add_to_hash
if ( q - > count = = 1 & & ! q - > callback_suspended & & q - > callback ) {
q - > callback ( q , q - > callback_arg ) ;
}
# ifdef QUEUE_DEBUG
// queue_check_consistency(q);// !!!! for debug - BEFORE callback
queue_check_consistency ( q ) ;
# endif
return 0 ;
}
int queue_data_put_with_index ( struct ll_queue * q , struct ll_entry * entry , uint16_t index_offset , uint16_t index_size ) {
if ( ! q | | ! entry ) return - 1 ;
# ifdef QUEUE_THREAD_CHECK
queue_check_thread ( q ) ;
# endif
if ( index_size = = 0 | | ( size_t ) index_offset + index_size > ( size_t ) entry - > size ) {
DEBUG_ERROR ( DEBUG_CATEGORY_LL_QUEUE , " [%s] invalid index: offset=%u size=%u (data_size=%u) " , q - > name , index_offset , index_size , entry - > size ) ;
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
entry - > index_offset = index_offset ;
entry - > index_size = index_size ;
entry - > index_hash = make_hash ( entry - > data + index_offset , index_size ) ;
entry - > id = id ;
// Проверить лимит размера
if ( q - > size_limit > = 0 & & q - > count > = q - > size_limit ) {
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ; // Освободить элемент если превышен лимит
queue_entry_free ( entry ) ;
return - 1 ;
}
// Добавить в конец
// Добавляем в конец
entry - > next = NULL ;
entry - > prev = q - > tail ;
if ( q - > tail ) {
q - > tail - > next = entry ;
} else {
q - > head = entry ;
}
if ( q - > tail ) q - > tail - > next = entry ; else q - > head = entry ;
q - > tail = entry ;
q - > count + + ;
entry - > int_len = entry - > len ;
entry - > int_len = entry - > len ;
q - > total_bytes + = entry - > int_len ;
size_t send_q_bytes = queue_total_bytes ( q ) ;
// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "check total bytes: new_q_len=%d element_size:%d", send_q_bytes, entry->size);
add_to_hash ( q , entry ) ;
// DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_data_put: added entry %p (id=%u), count=%d", entry, id, q->count);
// ВАЖНО: проверка консистентности ДО коллбэка, так как коллбэк может модифицировать очередь
if ( q - > count = = 1 & & ! q - > callback_suspended & & q - > callback ) {
q - > callback ( q , q - > callback_arg ) ;
}
# ifdef QUEUE_DEBUG
queue_check_consistency ( q ) ; // !!!! for debug - BEFORE callback
queue_check_consistency ( q ) ;
# endif
return 0 ;
}
int queue_data_put_first ( struct ll_queue * q , struct ll_entry * entry ) {
if ( ! q | | ! entry ) return - 1 ;
# ifdef QUEUE_THREAD_CHECK
queue_check_thread ( q ) ;
# endif
// Если очередь была пуста и коллбэки разрешены - вызвать коллбэк
entry - > index_offset = 0 ;
entry - > index_size = 0 ;
entry - > index_hash = 0 ;
if ( q - > hash_size > 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_LL_QUEUE , " [%s] Put no-hash data to hash queue " , q - > name ) ;
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
if ( q - > size_limit > = 0 & & q - > count > = q - > size_limit ) {
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ; // как было в оригинале
return - 1 ;
}
entry - > next = q - > head ;
entry - > prev = NULL ;
if ( q - > head ) q - > head - > prev = entry ; else q - > tail = entry ;
q - > head = entry ;
q - > count + + ;
entry - > int_len = entry - > len ;
q - > total_bytes + = entry - > int_len ;
if ( q - > count = = 1 & & ! q - > callback_suspended & & q - > callback ) {
q - > callback ( q , q - > callback_arg ) ;
}
// Проверить ожидающие коллбэки (надо только при заборе из очереди)
// check_waiters(q);
# ifdef QUEUE_DEBUG
queue_check_consistency ( q ) ; // !!!! for debug - AFTER callback
queue_check_consistency ( q ) ;
# endif
return 0 ;
}
int queue_data_put_first ( struct ll_queue * q , struct ll_entry * entry , uint32_t id ) {
// С хешем для put_first
int queue_data_put_first_with_index ( struct ll_queue * q , struct ll_entry * entry , uint16_t index_offset , uint16_t index_size ) {
if ( ! q | | ! entry ) return - 1 ;
# ifdef QUEUE_THREAD_CHECK
queue_check_thread ( q ) ;
# endif
# endif
if ( index_size = = 0 | | ( size_t ) index_offset + index_size > ( size_t ) entry - > size ) {
DEBUG_ERROR ( DEBUG_CATEGORY_LL_QUEUE , " [%s] invalid index: offset=%u size=%u (data_size=%u) " , q - > name , index_offset , index_size , entry - > size ) ;
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
entry - > index_offset = index_offset ;
entry - > index_size = index_size ;
entry - > index_hash = make_hash ( entry - > data + index_offset , index_size ) ;
entry - > id = id ;
// Проверить лимит размера
if ( q - > size_limit > = 0 & & q - > count > = q - > size_limit ) {
queue_entry_free ( entry ) ; // Освободить элемент если превышен лимит
queue_dgram_free ( entry ) ;
queue_entry_free ( entry ) ;
return - 1 ;
}
// Добавить в начало
entry - > next = q - > head ;
entry - > prev = NULL ;
if ( q - > head ) {
q - > head - > prev = entry ;
} else {
q - > tail = entry ;
}
if ( q - > head ) q - > head - > prev = entry ; else q - > tail = entry ;
q - > head = entry ;
q - > count + + ;
entry - > int_len = entry - > len ;
entry - > int_len = entry - > len ;
q - > total_bytes + = entry - > int_len ;
add_to_hash ( q , entry ) ;
// DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_data_put_first: added entry %p (id=%u), count=%d", entry, id, q->count);
// ВАЖНО: проверка консистентности ДО коллбэка
# ifdef QUEUE_DEBUG
queue_check_consistency ( q ) ; // !!!! for debug - BEFORE callback
# endif
// Если очередь была пуста и коллбэки разрешены - вызвать коллбэк
if ( q - > count = = 1 & & ! q - > callback_suspended & & q - > callback ) {
q - > callback ( q , q - > callback_arg ) ;
}
// Проверить ожидающие коллбэки
// check_waiters(q);
# ifdef QUEUE_DEBUG
queue_check_consistency ( q ) ; // !!!! for debug
queue_check_consistency ( q ) ;
# endif
return 0 ;
}
struct ll_entry * queue_data_get ( struct ll_queue * q ) {
if ( ! q | | ! q - > head ) return NULL ;
@ -496,19 +581,22 @@ void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter) {
// ==================== Поиск и удаление по ID ====================
struct ll_entry * queue_find_data_by_id ( struct ll_queue * q , uint32_t id ) {
if ( ! q | | q - > hash_size = = 0 | | ! q - > hash_table ) return NULL ;
size_t slot = id % q - > hash_size ;
struct ll_entry * queue_find_data_by_index ( struct ll_queue * q , const void * index_key , uint16_t index_size ) {
if ( ! q | | q - > hash_size = = 0 | | ! q - > hash_table | | index_size = = 0 | | ! index_key ) return NULL ;
uint32_t hash_val = make_hash ( index_key , index_size ) ;
uint32_t slot = hash_val % q - > hash_size ;
struct ll_entry * entry = q - > hash_table [ slot ] ;
while ( entry ) {
if ( entry - > id = = id ) {
if ( entry - > index_size = = index_size & &
( size_t ) entry - > index_offset + index_size < = ( size_t ) entry - > size & &
memcmp ( entry - > data + entry - > index_offset , index_key , index_size ) = = 0 ) {
return entry ;
}
entry = entry - > hash_next ;
}
return NULL ;
}
Evgeny
6 months ago