diff --git a/CMakeLists.txt b/CMakeLists.txt index 158ccdc..2a7c73c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -28,6 +28,7 @@ if (NOT NO_STATIC_ANALYSIS) set(cppcheck_options --enable=warning,style,performance,portability --error-exitcode=2 + -UCY_CONFIG_HEADER --suppress=missingIncludeSystem --suppress=badBitmaskCheck --suppress=constParameterCallback diff --git a/cy/cy.c b/cy/cy.c index 4bc9e2d..0217d97 100644 --- a/cy/cy.c +++ b/cy/cy.c @@ -17,11 +17,13 @@ #include "cy.h" #include "cy_platform.h" #include -#include // Configure cavl2.h. Key definitions must be provided before including the header. #define CAVL2_RELATION int32_t #define CAVL2_T cy_tree_t +#ifndef CAVL2_ASSERT +#define CAVL2_ASSERT(x) CY_ASSERT(x) +#endif typedef struct cy_tree_t cy_tree_t; struct cy_tree_t { @@ -39,7 +41,6 @@ struct cy_tree_t #include // Standard library includes come last. -#include #include #if CY_CONFIG_TRACE #include @@ -266,7 +267,7 @@ static void* wkv_realloc(wkv_t* const self, void* ptr, const size_t new_size) static void* mem_alloc(const cy_t* const cy, const size_t size) { - assert(cy != NULL); + CY_ASSERT(cy != NULL); return cy->platform->vtable->realloc(cy->platform, NULL, size); } @@ -281,7 +282,7 @@ static void* mem_alloc_zero(const cy_t* const cy, const size_t size) static void mem_free(const cy_t* const cy, void* ptr) { - assert(cy != NULL); + CY_ASSERT(cy != NULL); if (ptr != NULL) { cy->platform->vtable->realloc(cy->platform, ptr, 0); // NOLINT(*NullDereference) } @@ -297,10 +298,10 @@ static const cy_bytes_t bytes_empty_sentinel = { .size = 0, .data = "", .next = // Frees all memory allocated by bytes_dup(). No-op if bytes are NULL. static void bytes_undup(const cy_t* const cy, const cy_bytes_t* bytes) { - assert(cy != NULL); + CY_ASSERT(cy != NULL); if (&bytes_empty_sentinel != bytes) { while (bytes != NULL) { - assert(bytes->data == ((const void*)(bytes + 1))); + CY_ASSERT(bytes->data == ((const void*)(bytes + 1))); const cy_bytes_t* const next = bytes->next; mem_free(cy, (void*)bytes); bytes = next; @@ -311,7 +312,7 @@ static void bytes_undup(const cy_t* const cy, const cy_bytes_t* bytes) // Copies bytes to the heap in small chunks to reduce fragmentation risks. NULL iff OOM. Use bytes_undup() to undo. static const cy_bytes_t* bytes_dup(const cy_t* const cy, const cy_bytes_t src) { - assert(cy != NULL); + CY_ASSERT(cy != NULL); static const size_t data_per_chunk = BYTES_DUP_CHUNK - sizeof(cy_bytes_t); const cy_bytes_t* in = &src; size_t in_offset = 0; @@ -328,7 +329,7 @@ static const cy_bytes_t* bytes_dup(const cy_t* const cy, const cy_bytes_t src) } break; } - assert((in->size == 0) || (in->data != NULL)); + CY_ASSERT((in->size == 0) || (in->data != NULL)); cy_bytes_t* const chunk = (cy_bytes_t*)mem_alloc(cy, BYTES_DUP_CHUNK); if (chunk == NULL) { @@ -354,15 +355,15 @@ static const cy_bytes_t* bytes_dup(const cy_t* const cy, const cy_bytes_t src) if (in == NULL) { break; } - assert((in->size == 0) || (in->data != NULL)); + CY_ASSERT((in->size == 0) || (in->data != NULL)); const size_t copy_size = smaller(data_per_chunk - chunk->size, in->size - in_offset); - assert(copy_size > 0); + CY_ASSERT(copy_size > 0); memcpy(((byte_t*)(chunk + 1)) + chunk->size, ((const byte_t*)in->data) + in_offset, copy_size); chunk->size += copy_size; in_offset += copy_size; } - assert(chunk->size <= data_per_chunk); + CY_ASSERT(chunk->size <= data_per_chunk); } return head; } @@ -479,7 +480,7 @@ static size_t bitmap_clz(const bitmap_t* const bitmap, const size_t bit_count) if (bits != 0U) { const size_t first = 63U - clz64(bits & (0ULL - bits)); const size_t out = (i * 64U) + first; - assert(out < bit_count); + CY_ASSERT(out < bit_count); return out; } } @@ -493,7 +494,7 @@ static void bitmap_shift(bitmap_t* const bitmap, const size_t bit_count, const i { if ((bitmap != NULL) && (bit_count > 0U) && (shift_amount != 0)) { const size_t words = BITMAP_WORDS(bit_count); - assert(words > 0U); + CY_ASSERT(words > 0U); // Ignore non-existent bits in the tail word on input and output to prevent leakage. const size_t tail = bit_count % 64U; if (tail > 0U) { @@ -565,15 +566,15 @@ static void delist(cy_list_t* const list, cy_list_member_t* const member) } member->next = NULL; member->prev = NULL; - assert((list->head != NULL) == (list->tail != NULL)); + CY_ASSERT((list->head != NULL) == (list->tail != NULL)); } // If the item is already in the list, it will be delisted first. Can be used for moving to the front. static void enlist_head(cy_list_t* const list, cy_list_member_t* const member) { delist(list, member); - assert((member->next == NULL) && (member->prev == NULL)); - assert((list->head != NULL) == (list->tail != NULL)); + CY_ASSERT((member->next == NULL) && (member->prev == NULL)); + CY_ASSERT((list->head != NULL) == (list->tail != NULL)); member->next = list->head; if (list->head != NULL) { list->head->prev = member; @@ -582,7 +583,7 @@ static void enlist_head(cy_list_t* const list, cy_list_member_t* const member) if (list->tail == NULL) { list->tail = member; } - assert((list->head != NULL) && (list->tail != NULL)); + CY_ASSERT((list->head != NULL) && (list->tail != NULL)); } #define LIST_MEMBER(ptr, owner_type, owner_field) ((owner_type*)ptr_unbias((ptr), offsetof(owner_type, owner_field))) @@ -612,7 +613,7 @@ size_t cy_message_read(const cy_message_t* const msg, const size_t offset, const void cy_message_refcount_inc(cy_message_t* const msg) { if (msg != NULL) { - assert(msg->refcount > 0); + CY_ASSERT(msg->refcount > 0); msg->refcount++; } } @@ -620,7 +621,7 @@ void cy_message_refcount_inc(cy_message_t* const msg) void cy_message_refcount_dec(cy_message_t* const msg) { if ((msg != NULL) && (msg->vtable != NULL)) { - assert(msg->refcount > 0); + CY_ASSERT(msg->refcount > 0); msg->refcount--; if (msg->refcount == 0) { msg->vtable->destroy(msg); @@ -630,7 +631,7 @@ void cy_message_refcount_dec(cy_message_t* const msg) static void message_skip(cy_message_t* const msg, const size_t offset) { - assert((msg != NULL) && (msg->vtable != NULL) && (msg->vtable->skip != NULL)); + CY_ASSERT((msg != NULL) && (msg->vtable != NULL) && (msg->vtable->skip != NULL)); msg->vtable->skip(msg, offset); } @@ -670,9 +671,9 @@ struct cy_future_t static void* future_new(cy_t* const cy, const cy_future_vtable_t* const vtbl, const size_t derived_size) { - assert(derived_size >= sizeof(cy_future_t)); - assert(vtbl != NULL); - assert((vtbl->done != NULL) && (vtbl->error != NULL) && (vtbl->timeout != NULL) && (vtbl->dispose != NULL)); + CY_ASSERT(derived_size >= sizeof(cy_future_t)); + CY_ASSERT(vtbl != NULL); + CY_ASSERT((vtbl->done != NULL) && (vtbl->error != NULL) && (vtbl->timeout != NULL) && (vtbl->dispose != NULL)); cy_future_t* const future = (cy_future_t*)mem_alloc_zero(cy, derived_size); if (future != NULL) { future->index = TREE_NULL; @@ -714,7 +715,7 @@ static bool future_index_insert(cy_future_t* const self, cy_tree_t** const index // MUST be inserted, otherwise UB. static void future_index_remove(cy_future_t* const self, cy_tree_t** const index) { - assert(cavl2_is_inserted(*index, &self->index)); + CY_ASSERT(cavl2_is_inserted(*index, &self->index)); cavl2_remove(index, &self->index); } @@ -733,7 +734,7 @@ static bool future_indexed(const cy_future_t* const self, const cy_tree_t* const static void future_timeout_trampoline(olga_t* const sched, olga_event_t* const event, const cy_us_t now) { (void)sched; - assert(event->deadline <= now); + CY_ASSERT(event->deadline <= now); cy_future_t* const self = (cy_future_t*)event->user; self->vtable->timeout(self, event->deadline, now); } @@ -878,7 +879,7 @@ static writer_t* writer_acquire(cy_t* const cy, const uint32_t subject_id) // Decrement refcount; destroy and free when it reaches zero. static void writer_release(cy_t* const cy, writer_t* const w) { - assert((w != NULL) && (w->handle != NULL) && (w->refcount > 0)); + CY_ASSERT((w != NULL) && (w->handle != NULL) && (w->refcount > 0)); w->refcount--; if (w->refcount == 0) { cavl2_remove(&cy->writers, &w->index); @@ -906,7 +907,7 @@ static reader_t* reader_acquire(cy_t* const cy, const uint32_t subject_id, const static void reader_release(cy_t* const cy, reader_t* const r) { - assert((r != NULL) && (r->handle != NULL) && (r->refcount > 0)); + CY_ASSERT((r != NULL) && (r->handle != NULL) && (r->refcount > 0)); r->refcount--; if (r->refcount == 0) { cavl2_remove(&cy->readers, &r->index); @@ -918,7 +919,7 @@ static void reader_release(cy_t* const cy, reader_t* const r) // No-op if the current extent is sufficient. static void reader_grow_extent(cy_t* const cy, reader_t* const r, const size_t new_extent) { - assert((r != NULL) && (r->handle != NULL) && (r->refcount > 0)); + CY_ASSERT((r != NULL) && (r->handle != NULL) && (r->refcount > 0)); if (new_extent > r->handle->extent) { cy->platform->vtable->subject_reader_extent_set(cy->platform, r->handle, new_extent); r->handle->extent = new_extent; @@ -1033,7 +1034,7 @@ typedef struct } topic_repr_t; static topic_repr_t topic_repr(const cy_topic_t* const topic) { - assert(topic != NULL); + CY_ASSERT(topic != NULL); topic_repr_t out = { 0 }; const cy_str_t name = cy_topic_name(topic); (void)snprintf(out.str, @@ -1109,7 +1110,7 @@ static int32_t association_cavl_compare(const void* const user, const cy_tree_t* static void association_forget(cy_topic_t* const topic, association_t* const assoc) { cy_t* const cy = topic->cy; - assert(cavl2_is_inserted(topic->assoc_by_remote_id, &assoc->index_remote_id)); + CY_ASSERT(cavl2_is_inserted(topic->assoc_by_remote_id, &assoc->index_remote_id)); cavl2_remove(&topic->assoc_by_remote_id, &assoc->index_remote_id); topic->assoc_count--; CY_TRACE(cy, "⛓🗑 %s N%016jx count=%zu", topic_repr(topic).str, (uintmax_t)assoc->remote_id, topic->assoc_count); @@ -1140,12 +1141,12 @@ static cy_tree_t* association_cavl_factory(void* const user) // the given remote-ID or the position where it should be inserted if not found. static size_t association_bisect(association_t* const* const assoc, const size_t count, const uint64_t remote_id) { - assert((assoc != NULL) || (count == 0)); + CY_ASSERT((assoc != NULL) || (count == 0)); size_t lo = 0; size_t hi = count; while (lo < hi) { const size_t mid = lo + ((hi - lo) / 2U); - assert(assoc[mid] != NULL); + CY_ASSERT(assoc[mid] != NULL); if (assoc[mid]->remote_id < remote_id) { lo = mid + 1U; } else { @@ -1223,7 +1224,7 @@ static void topic_merge_lage(cy_topic_t* const topic, const cy_us_t now, int_fas // This comparator is only applicable on subject-ID allocation conflicts. As such, hashes must be different. static bool left_wins(const cy_topic_t* const left, const cy_us_t now, const int_fast8_t r_lage, const uint64_t r_hash) { - assert(left->hash != r_hash); + CY_ASSERT(left->hash != r_hash); const int_fast8_t l_lage = topic_lage(left, now); return (l_lage != r_lage) ? (l_lage > r_lage) : left->hash < r_hash; // older topic wins } @@ -1271,7 +1272,7 @@ static void topic_sync_implicit(cy_topic_t* const topic) CY_TRACE(topic->cy, "🧛 %s promoted to explicit", topic_repr(topic).str); } } - assert(implicit == is_implicit(topic)); + CY_ASSERT(implicit == is_implicit(topic)); } // Move the topic to the head of the doubly-linked list of implicit topics. @@ -1289,7 +1290,7 @@ static void retire_expired_implicit_topics(cy_t* const cy, const cy_us_t now) { cy_topic_t* const topic = LIST_TAIL(cy->list_implicit, cy_topic_t, list_implicit); if (topic != NULL) { - assert(is_implicit(topic) && topic_validate_is_implicit(topic)); + CY_ASSERT(is_implicit(topic) && topic_validate_is_implicit(topic)); if ((topic->ts_animated + cy->implicit_topic_timeout) < now) { CY_TRACE(cy, "⚰️ %s", topic_repr(topic).str); // Expiration may occur while pattern subscribers are still alive, so we have to detach @@ -1308,20 +1309,20 @@ static uint32_t topic_subject_id_impl(const uint64_t hash, const uint64_t evicti // Pinned: subject-ID = UINT32_MAX - evictions, in [0, CY_SUBJECT_ID_PINNED_MAX]. return (uint32_t)(UINT32_MAX - (uint32_t)evictions); } - assert(subject_id_modulus > 0); - assert(evictions <= UINT32_MAX); + CY_ASSERT(subject_id_modulus > 0); + CY_ASSERT(evictions <= UINT32_MAX); const uint64_t h = hash % subject_id_modulus; const uint64_t e = evictions % subject_id_modulus; const uint64_t subject_id = CY_SUBJECT_ID_PINNED_MAX + 1ULL + ((h + ((e * e) % subject_id_modulus)) % subject_id_modulus); - assert((subject_id > CY_SUBJECT_ID_PINNED_MAX) && (subject_id <= CY_SUBJECT_ID_MAX(subject_id_modulus))); - assert(subject_id <= UINT32_MAX); + CY_ASSERT((subject_id > CY_SUBJECT_ID_PINNED_MAX) && (subject_id <= CY_SUBJECT_ID_MAX(subject_id_modulus))); + CY_ASSERT(subject_id <= UINT32_MAX); return (uint32_t)subject_id; } static uint32_t topic_subject_id(const cy_topic_t* const topic) { - assert(topic != NULL); + CY_ASSERT(topic != NULL); return topic_subject_id_impl(topic->hash, topic->evictions, topic->cy->platform->subject_id_modulus); } @@ -1330,9 +1331,9 @@ static uint32_t topic_gossip_shard_subject_id(const cy_t* const cy, const uint64 // Gossip shard subjects are located between the max valid subject-ID and the broadcast subject-ID. const uint32_t shard_index = (uint32_t)(topic_hash % (uint64_t)cy->gossip_shard_count); const uint32_t subject_id = CY_SUBJECT_ID_MAX(cy->platform->subject_id_modulus) + 1U + shard_index; - assert(subject_id > CY_SUBJECT_ID_MAX(cy->platform->subject_id_modulus)); - assert(subject_id < cy->broad_reader->subject_id); - assert(cy->broad_reader->subject_id == cy->broad_writer->subject_id); + CY_ASSERT(subject_id > CY_SUBJECT_ID_MAX(cy->platform->subject_id_modulus)); + CY_ASSERT(subject_id < cy->broad_reader->subject_id); + CY_ASSERT(cy->broad_reader->subject_id == cy->broad_writer->subject_id); return subject_id; } @@ -1342,7 +1343,7 @@ static cy_topic_t* topic_find_by_subject_id(const cy_t* const cy, const uint32_t { cy_topic_t* const topic = CAVL2_TO_OWNER( cavl2_find(cy->topics_by_subject_id, &subject_id, &cavl_comp_topic_subject_id), cy_topic_t, index_subject_id); - assert((topic == NULL) || (topic_subject_id(topic) == subject_id)); + CY_ASSERT((topic == NULL) || (topic_subject_id(topic) == subject_id)); return topic; } @@ -1370,7 +1371,7 @@ static void topic_sync_subject_reader(cy_topic_t* const topic) const uint32_t subject_id = topic_subject_id(topic); if ((topic->couplings != NULL) && (topic->sub_reader == NULL)) { // A subject reader is needed but missing! const size_t extent = subscription_extent_w_overhead(topic); - assert(extent >= HEADER_BYTES); + CY_ASSERT(extent >= HEADER_BYTES); topic->sub_reader = reader_acquire(cy, subject_id, extent); CY_TRACE(topic->cy, "🗞️ %s S%08jx extent=%zu result=%p", @@ -1382,7 +1383,7 @@ static void topic_sync_subject_reader(cy_topic_t* const topic) ON_ASYNC_ERROR(cy, topic, CY_ERR_MEMORY); } } - assert((topic->sub_reader == NULL) || (topic->sub_reader->handle->subject_id == subject_id)); + CY_ASSERT((topic->sub_reader == NULL) || (topic->sub_reader->handle->subject_id == subject_id)); if ((topic->couplings == NULL) && (topic->sub_reader != NULL)) { // No longer needed. reader_release(cy, topic->sub_reader); topic->sub_reader = NULL; @@ -1448,7 +1449,7 @@ static void topic_allocate(cy_topic_t* const topic, const uint32_t new_evictions const uint32_t new_sid = topic_subject_id_impl(topic->hash, new_evictions, cy->platform->subject_id_modulus); cy_topic_t* const that = CAVL2_TO_OWNER( cavl2_find(cy->topics_by_subject_id, &new_sid, &cavl_comp_topic_subject_id), cy_topic_t, index_subject_id); - assert((that == NULL) || (topic->hash != that->hash)); // This would mean that we inserted the same topic twice + CY_ASSERT((that == NULL) || (topic->hash != that->hash)); // This would mean that we inserted the same topic twice const bool same_subject = new_sid == old_sid; const bool victory = (that == NULL) || left_wins(topic, now, topic_lage(that, now), that->hash); @@ -1466,10 +1467,10 @@ static void topic_allocate(cy_topic_t* const topic, const uint32_t new_evictions if (victory) { // Allocation done. Every affected topic will end up here eventually. // Release old handles only for the subject we're leaving. - assert((topic->sub_reader == NULL) || (topic->sub_reader->handle->subject_id == old_sid)); - assert((topic->pub_writer == NULL) || (topic->pub_writer->handle->subject_id == old_sid)); + CY_ASSERT((topic->sub_reader == NULL) || (topic->sub_reader->handle->subject_id == old_sid)); + CY_ASSERT((topic->pub_writer == NULL) || (topic->pub_writer->handle->subject_id == old_sid)); if (!same_subject && (topic->sub_reader != NULL)) { - assert(topic->couplings != NULL); + CY_ASSERT(topic->couplings != NULL); reader_release(cy, topic->sub_reader); topic->sub_reader = NULL; } @@ -1480,7 +1481,7 @@ static void topic_allocate(cy_topic_t* const topic, const uint32_t new_evictions if (that != NULL) { cavl2_remove(&cy->topics_by_subject_id, &that->index_subject_id); - assert(topic->pub_writer == NULL); + CY_ASSERT(topic->pub_writer == NULL); topic->pub_writer = that->pub_writer; // Preserve transport state for the winning subject, if any. that->pub_writer = NULL; } @@ -1494,7 +1495,7 @@ static void topic_allocate(cy_topic_t* const topic, const uint32_t new_evictions &cavl2_trivial_factory), cy_topic_t, index_subject_id); - assert(self == topic); + CY_ASSERT(self == topic); (void)self; // The subject reader, if needed, must be acquired eagerly from the registry. @@ -1681,7 +1682,7 @@ static cy_err_t topic_couple(cy_topic_t* const topic, // string, which is part of the topic object. const wkv_substitution_t* s = substitutions; for (size_t i = 0U; s != NULL; i++) { - assert(i < cpl->substitution_count); + CY_ASSERT(i < cpl->substitution_count); cpl->substitutions[i] = (cy_substitution_t){ .str = s->str, .ordinal = s->ordinal }; s = s->next; } @@ -1734,7 +1735,7 @@ static cy_topic_t* topic_subscribe_if_matching(cy_t* const cy, const uint32_t evictions, const int_fast8_t lage) { - assert((cy != NULL) && (resolved_name.str != NULL)); + CY_ASSERT((cy != NULL) && (resolved_name.str != NULL)); if ((resolved_name.len == 0) || (name_normalized_len(resolved_name) != resolved_name.len) || !name_is_verbatim(resolved_name) || (rapidhash(resolved_name.str, resolved_name.len) != hash)) { return NULL; // Ensure the remote is not trying to feed us a bad name. @@ -1856,7 +1857,7 @@ static void gossip_event_urgent(olga_t* const olga, olga_event_t* const event, c (void)olga; cy_topic_t* const topic = (cy_topic_t*)event->user; schedule_gossip_periodic(topic, now, false); - assert(topic->cy->gossip_broadcast_ratio > 0); + CY_ASSERT(topic->cy->gossip_broadcast_ratio > 0); topic->gossip_counter = 0; ON_ASYNC_ERROR_IF(topic->cy, topic, send_gossip_multicast(topic, now, topic->cy->broad_writer)); } @@ -1907,7 +1908,7 @@ static void schedule_gossip_urgent(cy_topic_t* const topic, const cy_us_t now) // The bottom-level scout transmission function. Scouts are always broadcast. static cy_err_t do_send_scout(const cy_t* const cy, const cy_us_t now, const cy_str_t pattern) { - assert(pattern.len <= CY_TOPIC_NAME_MAX); + CY_ASSERT(pattern.len <= CY_TOPIC_NAME_MAX); byte_t buf[HEADER_BYTES + CY_TOPIC_NAME_MAX]; buf[0] = header_scout; memset(&buf[1], 0, HEADER_BYTES - 2); @@ -1965,7 +1966,7 @@ static void on_gossip_known_topic(cy_t* const cy, schedule_gossip_urgent(mine, ts); } else { topic_allocate(mine, evictions, ts); - assert(mine->gossip_event.handler == gossip_event_urgent); + CY_ASSERT(mine->gossip_event.handler == gossip_event_urgent); if (mine->evictions == evictions) { // no need to urgent-gossip: subject occupancy has not been altered schedule_gossip_periodic(mine, ts, true); // cancel urgent } @@ -1998,7 +1999,7 @@ static void on_gossip_unknown_topic(cy_t* const cy, if (mine == NULL) { return; // We are not using this subject-ID, no collision. } - assert(topic_subject_id(mine) == subject_id); + CY_ASSERT(topic_subject_id(mine) == subject_id); const bool win = left_wins(mine, ts, lage, hash); CY_TRACE(cy, "💥 Collision on S%08jx:\n" @@ -2109,10 +2110,10 @@ cy_publisher_t* cy_advertise_client(cy_t* const cy, const cy_str_t name, const s pub->priority = cy_prio_nominal; pub->ack_baseline_timeout = cy->ack_baseline_timeout; if (res == CY_OK) { - assert(pub->topic != NULL); + CY_ASSERT(pub->topic != NULL); pub->topic->pub_count++; topic_sync_implicit(pub->topic); - assert(!is_implicit(pub->topic)); + CY_ASSERT(!is_implicit(pub->topic)); const size_t response_extent_with_header = response_extent + HEADER_BYTES; if (response_extent_with_header > cy->unicast_extent) { // Currently, we only increase the extent and leave it at the max. Ideally we should also shrink it when @@ -2145,7 +2146,7 @@ static cy_err_t do_publish_impl(cy_publisher_t* const pub, cy_topic_t* const topic = pub->topic; cy_t* const cy = topic->cy; const cy_platform_vtable_t* const vt = cy->platform->vtable; - assert(topic->pub_count > 0); + CY_ASSERT(topic->pub_count > 0); byte_t header[HEADER_BYTES] = { (byte_t)header_type, 0, 0, (byte_t)topic_lage(topic, cy_now(cy)) }; (void)serialize_u32(&header[4], topic->evictions); @@ -2166,8 +2167,8 @@ static cy_err_t do_publish_impl(cy_publisher_t* const pub, if (err != CY_OK) { return err; } - assert(topic->pub_writer != NULL); - assert(topic_subject_id(topic) == topic->pub_writer->handle->subject_id); + CY_ASSERT(topic->pub_writer != NULL); + CY_ASSERT(topic_subject_id(topic) == topic->pub_writer->handle->subject_id); return vt->subject_writer_send(cy->platform, topic->pub_writer->handle, deadline, pub->priority, headed_message); } @@ -2181,7 +2182,7 @@ static cy_err_t do_publish(cy_publisher_t* const pub, return CY_ERR_ARGUMENT; } cy_topic_t* const topic = pub->topic; - assert(topic->pub_count > 0); + CY_ASSERT(topic->pub_count > 0); const uint64_t tag = topic->pub_tag_baseline + topic->pub_seqno++; if (out_tag != NULL) { *out_tag = tag; @@ -2235,14 +2236,14 @@ static void publish_future_release_associations(publish_future_t* const self) // Don't register ack loss when canceled prematurely because there may not have been enough time for round trip. // Also, don't register any ack loss if at least one publication has failed, that is obvious. const uint64_t seqno = self->base.key - topic->pub_tag_baseline; - assert(seqno < (1ULL << 48U)); // sanity/math check -- values above 2**48 are unreachable in practice. + CY_ASSERT(seqno < (1ULL << 48U)); // sanity/math check -- values above 2**48 are unreachable in practice. if (bitmap_test(self->assoc_knockout, i) && (seqno >= ass->seqno_witness) && (!self->compromised)) { ass->slack++; } // Decrement refcount and remove the association if the slack is too large. // If refcount is not zero, it is someone else's responsibility to remove it (unless revived by then). - assert(ass->pending_count > 0); + CY_ASSERT(ass->pending_count > 0); ass->pending_count--; if ((ass->slack >= topic->assoc_slack_limit) && (ass->pending_count == 0)) { association_forget(topic, ass); @@ -2257,7 +2258,7 @@ static void publish_future_release_associations(publish_future_t* const self) // Invalidates the future -- the user callback may destroy it. Expect finalization & further access invalid. static void publish_future_materialize(publish_future_t* const self, const cy_err_t error) { - assert(!self->done); + CY_ASSERT(!self->done); self->done = true; const cy_t* const cy = self->owner->topic->cy; self->error = error; @@ -2292,7 +2293,7 @@ static bool ack_is_last_attempt(const cy_us_t current_ack_deadline, static void publish_future_timeout(cy_future_t* const base, const cy_us_t scheduled, const cy_us_t now) { - assert(scheduled <= now); // scheduler invariant + CY_ASSERT(scheduled <= now); // scheduler invariant (void)scheduled; publish_future_t* const self = (publish_future_t*)base; cy_topic_t* const topic = self->owner->topic; @@ -2306,18 +2307,18 @@ static void publish_future_timeout(cy_future_t* const base, const cy_us_t schedu self->error = sched_lag_error ? CY_ERR_LAG : self->error; // Weak error, may be overwritten below. // Check completion. - assert(!self->done); + CY_ASSERT(!self->done); if ((self->data == NULL) || (now >= self->deadline)) { // This is the final poll. publish_future_materialize(self, self->acknowledged ? CY_OK : CY_ERR_DELIVERY); return; } // Compute next deadline and decide if it's going to be the last attempt based on the remaining time. - assert(now < self->deadline); + CY_ASSERT(now < self->deadline); self->ack_timeout *= 2; // exponential backoff const cy_us_t ack_deadline = sooner(self->ack_timeout + now, self->deadline); // manage possible scheduler lag const bool last_attempt = ack_is_last_attempt(ack_deadline, self->ack_timeout, self->deadline); - assert(ack_deadline > now); + CY_ASSERT(ack_deadline > now); // We can use multicast throughout, but it may be inefficient if we only need to reach few remaining subscribers. // This is not a correctness issue because each subscriber will receive our message at most once per attempt, @@ -2334,10 +2335,10 @@ static void publish_future_timeout(cy_future_t* const base, const cy_us_t schedu const cy_lane_t* lane_p = NULL; if (unicast) { const size_t assoc_idx = bitmap_clz(self->assoc_knockout, self->assoc_capacity); - assert(assoc_idx < self->assoc_capacity); - assert(bitmap_test(self->assoc_knockout, assoc_idx)); + CY_ASSERT(assoc_idx < self->assoc_capacity); + CY_ASSERT(bitmap_test(self->assoc_knockout, assoc_idx)); const association_t* const assoc = self->assoc_set[assoc_idx]; - assert(assoc != NULL); + CY_ASSERT(assoc != NULL); lane = (cy_lane_t){ .id = assoc->remote_id, .ctx = assoc->unicast_ctx, .prio = self->owner->priority }; CY_TRACE(cy, "☝️ %s Unicast to N%016jx tag=%016jx", @@ -2362,13 +2363,13 @@ static void publish_future_timeout(cy_future_t* const base, const cy_us_t schedu self->data = NULL; future_deadline_arm(base, self->deadline); } else { - assert(ack_deadline < self->deadline); + CY_ASSERT(ack_deadline < self->deadline); future_deadline_arm(base, ack_deadline); } // Notify if any errors occurred, even if not yet done. The user will check the done state. if ((er != CY_OK) || sched_lag_error) { - assert(self->error != CY_OK); + CY_ASSERT(self->error != CY_OK); future_notify(&self->base); // Invalidates the future. Expect disposal. } } @@ -2380,12 +2381,12 @@ static void publish_future_dispose(cy_future_t* const base) self->compromised = true; // Prevent slack adjustment because we're disposing early. publish_future_materialize(self, CY_OK); } - assert(self->done); - assert(self->assoc_capacity == 0); - assert(self->assoc_knockout == NULL); - assert(self->data == NULL); - assert(!future_deadline_armed(base)); - assert(!future_indexed(base, self->owner->topic->pub_futures_by_tag)); + CY_ASSERT(self->done); + CY_ASSERT(self->assoc_capacity == 0); + CY_ASSERT(self->assoc_knockout == NULL); + CY_ASSERT(self->data == NULL); + CY_ASSERT(!future_deadline_armed(base)); + CY_ASSERT(!future_indexed(base, self->owner->topic->pub_futures_by_tag)); mem_free(base->cy, self); } @@ -2396,14 +2397,14 @@ static const cy_future_vtable_t publish_future_vtable = { .done = publish_fut static void publish_future_on_ack(publish_future_t* const self, const uint64_t remote_id, const bool positive) { - assert(!self->done); + CY_ASSERT(!self->done); // Regardless of whether it's a positive ack or a nack, we need to cease further deliveries. const size_t idx = association_bisect(self->assoc_set, self->assoc_capacity, remote_id); const bool known = (idx < self->assoc_capacity) && (self->assoc_set[idx]->remote_id == remote_id); if (known && bitmap_test(self->assoc_knockout, idx)) { - assert(self->assoc_set[idx]->pending_count > 0); + CY_ASSERT(self->assoc_set[idx]->pending_count > 0); bitmap_clear(self->assoc_knockout, idx); - assert(self->assoc_remaining > 0); + CY_ASSERT(self->assoc_remaining > 0); self->assoc_remaining--; } self->acknowledged = self->acknowledged || positive; // optimistic success based on a single +ack @@ -2473,7 +2474,7 @@ cy_future_t* cy_publish_reliable(cy_publisher_t* const pub, const cy_us_t deadli // Populate the association pointer array ORDERED from low to high IDs for fast lookups. association_t* ass_cursor = (association_t*)cavl2_min(topic->assoc_by_remote_id); while ((fut->assoc_remaining < fut->assoc_capacity) && (ass_cursor != NULL)) { - assert(fut->assoc_knockout != NULL); + CY_ASSERT(fut->assoc_knockout != NULL); // Some associations may be pending removal already, skip them. There will be unused pointers but we don't care. if (ass_cursor->slack < topic->assoc_slack_limit) { bitmap_set(fut->assoc_knockout, fut->assoc_remaining); @@ -2481,7 +2482,7 @@ cy_future_t* cy_publish_reliable(cy_publisher_t* const pub, const cy_us_t deadli fut->assoc_remaining++; ass_cursor->pending_count++; } else { - assert(ass_cursor->pending_count > 0); // Sanity check -- otherwise would have been removed. + CY_ASSERT(ass_cursor->pending_count > 0); // Sanity check -- otherwise would have been removed. } ass_cursor = (association_t*)cavl2_next_greater((cy_tree_t*)ass_cursor); } @@ -2489,7 +2490,7 @@ cy_future_t* cy_publish_reliable(cy_publisher_t* const pub, const cy_us_t deadli // Complete the final infallible steps. const bool insert_ok = future_index_insert(&fut->base, &topic->pub_futures_by_tag, fut->base.key); - assert(insert_ok); // cannot fail by design + CY_ASSERT(insert_ok); // cannot fail by design (void)insert_ok; future_deadline_arm(&fut->base, one_shot ? deadline : ack_deadline); return &fut->base; @@ -2557,8 +2558,8 @@ static cy_tree_t* request_future_remote_cavl_factory(void* const user) static void request_publish_callback(cy_future_t* const fut) { request_future_t* const self = (request_future_t*)cy_future_context(fut).ptr[0]; - assert(self->publish == fut); - assert(!self->finalized); + CY_ASSERT(self->publish == fut); + CY_ASSERT(!self->finalized); const cy_err_t err = cy_future_error(fut); if (cy_future_done(fut)) { // In case there are intermediate updates. May be uncoverable. cy_future_destroy(fut); @@ -2587,9 +2588,9 @@ static response_rx_t request_on_response(request_future_t* const self, const bool reliable, const cy_lane_t lane) { - assert(seqno <= SEQNO48_MASK); - assert(message.timestamp >= 0); - assert(message.content != NULL); + CY_ASSERT(seqno <= SEQNO48_MASK); + CY_ASSERT(message.timestamp >= 0); + CY_ASSERT(message.content != NULL); cy_t* const cy = self->base.cy; // Zombie mode -- the application has destroyed the future and is no longer accepting responses. @@ -2635,7 +2636,7 @@ static response_rx_t request_on_response(request_future_t* const self, } bitmap_set(remote->seqno_acked, (size_t)dist); // genuinely new response just arrived out of order } - assert(remote->seqno_top >= seqno); + CY_ASSERT(remote->seqno_top >= seqno); } // At this point, the response is known to be unique. Rewrite the last stored response. @@ -2658,8 +2659,8 @@ static response_rx_t request_on_response(request_future_t* const self, static void request_future_destroy(request_future_t* const self) { cy_future_t* const base = &self->base; - assert(self->finalized); - assert(self->publish == NULL); + CY_ASSERT(self->finalized); + CY_ASSERT(self->publish == NULL); future_deadline_disarm(base); cy_message_refcount_dec(self->last_response.message.content); // NULL-safe future_index_remove(base, &self->topic->request_futures_by_tag); @@ -2674,7 +2675,7 @@ static void request_future_destroy(request_future_t* const self) static bool request_future_done(const cy_future_t* const base) { const request_future_t* const self = (const request_future_t*)base; - assert(!self->finalized); // use after free? + CY_ASSERT(!self->finalized); // use after free? return (self->last_response.message.content != NULL) || !future_deadline_armed(base); // got response or timed out } static cy_err_t request_future_error(const cy_future_t* const base) { return ((const request_future_t*)base)->error; } @@ -2684,7 +2685,7 @@ static void request_future_timeout(cy_future_t* const base, const cy_us_t schedu (void)scheduled; (void)now; request_future_t* const self = (request_future_t*)base; - assert(!future_deadline_armed(base)); + CY_ASSERT(!future_deadline_armed(base)); if (!self->finalized) { self->error = CY_ERR_LIVENESS; future_notify(base); // Expect finalization call. @@ -2696,7 +2697,7 @@ static void request_future_timeout(cy_future_t* const base, const cy_us_t schedu static void request_future_dispose(cy_future_t* const base) { request_future_t* const self = (request_future_t*)base; - assert(!self->finalized); + CY_ASSERT(!self->finalized); if (self->publish != NULL) { cy_future_destroy(self->publish); self->publish = NULL; @@ -2752,7 +2753,7 @@ cy_future_t* cy_request(cy_publisher_t* const pub, // Set up our future; this is infallible. Use the same tag for response correlation. const bool insert_ok = future_index_insert(&fut->base, &topic->request_futures_by_tag, fut->publish->key); - assert(insert_ok); // cannot fail by design, tags are per-topic unique + CY_ASSERT(insert_ok); // cannot fail by design, tags are per-topic unique (void)insert_ok; future_deadline_arm(&fut->base, delivery_deadline + response_timeout); @@ -2807,7 +2808,7 @@ void cy_priority_set(cy_publisher_t* const pub, const cy_prio_t priority) static cy_us_t derive_ack_timeout(const cy_us_t ack_baseline_timeout, const cy_prio_t priority) { - assert(ack_baseline_timeout > 0); + CY_ASSERT(ack_baseline_timeout > 0); return ack_baseline_timeout * (1LL << (byte_t)priority); // NOLINT(*signed*) } @@ -2835,10 +2836,10 @@ void cy_unadvertise(cy_publisher_t* const pub) cy_topic_t* const topic = pub->topic; // Dereference the topic. - assert(!is_implicit(topic)); - assert(topic->pub_count > 0); + CY_ASSERT(!is_implicit(topic)); + CY_ASSERT(topic->pub_count > 0); topic->pub_count--; - assert(!is_implicit(pub->topic)); + CY_ASSERT(!is_implicit(pub->topic)); topic_sync_implicit(topic); // topics are destroyed lazily via garbage collection to avoid state loss // Bye bye. @@ -2904,7 +2905,7 @@ static void subscriber_timeout(cy_future_t* const base, const cy_us_t scheduled, (void)scheduled; (void)now; subscriber_t* const self = (subscriber_t*)base; - assert((self->root != NULL) && (self->root->cy == base->cy)); + CY_ASSERT((self->root != NULL) && (self->root->cy == base->cy)); if (!self->disposed) { subscriber_notify_error(self, CY_ERR_LIVENESS); } else { @@ -2915,7 +2916,7 @@ static void subscriber_timeout(cy_future_t* const base, const cy_us_t scheduled, static void subscriber_dispose(cy_future_t* const base) { subscriber_t* const self = (subscriber_t*)base; - assert(!self->disposed); // use after free + CY_ASSERT(!self->disposed); // use after free #if CY_CONFIG_TRACE char name[CY_TOPIC_NAME_MAX + 1]; cy_subscriber_name(base, name); @@ -2962,7 +2963,7 @@ static bool subscriber_notify(subscriber_t* const self, const cy_arrival_t arriv if (self->params.liveness_timeout > 0) { future_deadline_arm(&self->base, arrival.message.timestamp + self->params.liveness_timeout); } - assert(self->base.vtable->done(&self->base)); + CY_ASSERT(self->base.vtable->done(&self->base)); subscriber_notify_error(self, CY_OK); return true; } @@ -3026,7 +3027,7 @@ static void dedup_commit(dedup_t* const self, const uint64_t tag) const uint64_t fwd = tag - self->tag; // Wrapping arithmetic. const uint64_t rev = self->tag - tag; // Wrapping arithmetic. if (rev < DEDUP_HISTORY) { // Out-of-order but within the window. - assert(!bitmap_test(self->bitmap, (size_t)rev)); + CY_ASSERT(!bitmap_test(self->bitmap, (size_t)rev)); bitmap_set(self->bitmap, (size_t)rev); } else { // Push the frontier or reset. if (fwd < DEDUP_HISTORY) { @@ -3042,7 +3043,7 @@ static void dedup_commit(dedup_t* const self, const uint64_t tag) static void dedup_destroy(dedup_t* const self, cy_topic_t* const owner) { delist(&owner->sub_list_dedup_by_recency, &self->list_recency); - assert(cavl2_is_inserted(owner->sub_index_dedup_by_remote_id, &self->index_remote_id)); + CY_ASSERT(cavl2_is_inserted(owner->sub_index_dedup_by_remote_id, &self->index_remote_id)); cavl2_remove(&owner->sub_index_dedup_by_remote_id, &self->index_remote_id); mem_free(owner->cy, self); } @@ -3153,26 +3154,26 @@ typedef struct // Remove the slot and invoke the user callback. static void reordering_eject(reordering_t* const self, reordering_slot_t* const slot) { - assert(slot != NULL); - assert(self->topic->cy == self->subscriber->root->cy); + CY_ASSERT(slot != NULL); + CY_ASSERT(self->topic->cy == self->subscriber->root->cy); const cy_t* const cy = self->topic->cy; // Remove the slot from the index. - assert(cavl2_is_inserted(self->interned_by_lin_tag, &slot->index_lin_tag)); + CY_ASSERT(cavl2_is_inserted(self->interned_by_lin_tag, &slot->index_lin_tag)); cavl2_remove(&self->interned_by_lin_tag, &slot->index_lin_tag); - assert(self->interned_count > 0); + CY_ASSERT(self->interned_count > 0); self->interned_count--; - assert((self->interned_by_lin_tag == NULL) == (self->interned_count == 0)); + CY_ASSERT((self->interned_by_lin_tag == NULL) == (self->interned_count == 0)); // Update the state with the removed slot. - assert(slot->lin_tag < (1ULL << 48U)); // ensure linearized by comparing against some unreachable value - assert(self->subscriber->params.reordering_window >= 0); // we should only end up here if ordered mode is used - assert(slot->lin_tag > self->last_ejected_lin_tag); // ensure ordered sequence seen by the application + CY_ASSERT(slot->lin_tag < (1ULL << 48U)); // ensure linearized by comparing against some unreachable value + CY_ASSERT(self->subscriber->params.reordering_window >= 0); // we should only end up here if ordered mode is used + CY_ASSERT(slot->lin_tag > self->last_ejected_lin_tag); // ensure ordered sequence seen by the application self->last_ejected_lin_tag = slot->lin_tag; // Construct the arrival instance. It copies the relevant states from the slot so that it can be destroyed. - assert(slot->message.timestamp >= 0); - assert(slot->message.content != NULL); + CY_ASSERT(slot->message.timestamp >= 0); + CY_ASSERT(slot->message.content != NULL); const cy_arrival_t arrival = make_arrival(self->topic, (cy_lane_t){ .id = self->remote_id, .ctx = self->unicast_ctx, .prio = slot->priority }, @@ -3219,7 +3220,7 @@ static void reordering_eject_all(reordering_t* const self, const bool silenced) while (self->interned_count > 0) { reordering_slot_t* const slot = CAVL2_TO_OWNER(cavl2_min(self->interned_by_lin_tag), reordering_slot_t, index_lin_tag); - assert((slot != NULL) && cavl2_is_inserted(self->interned_by_lin_tag, &slot->index_lin_tag)); + CY_ASSERT((slot != NULL) && cavl2_is_inserted(self->interned_by_lin_tag, &slot->index_lin_tag)); if (!silenced) { reordering_eject(self, slot); } else { @@ -3229,8 +3230,8 @@ static void reordering_eject_all(reordering_t* const self, const bool silenced) mem_free(self->topic->cy, slot); } } - assert(self->interned_count == 0); - assert(self->interned_by_lin_tag == NULL); + CY_ASSERT(self->interned_count == 0); + CY_ASSERT(self->interned_by_lin_tag == NULL); olga_cancel(&self->topic->cy->olga, &self->timeout); } @@ -3239,8 +3240,8 @@ static void reordering_resequence(reordering_t* const self, const uint64_t tag) // We do NOT accept the message immediately because we don't know if it's in order or not, as we don't have state. // For example, if we receive tag 3, we don't know if it's in a sequence of (3 2 1) or (3 4 5); to properly // handle the former case without message loss we start with the reordering delay. - assert(self->interned_count == 0); - assert(self->interned_by_lin_tag == NULL); + CY_ASSERT(self->interned_count == 0); + CY_ASSERT(self->interned_by_lin_tag == NULL); self->tag_baseline = tag - (REORDERING_CAPACITY / 2U); self->last_ejected_lin_tag = 0; } @@ -3254,8 +3255,8 @@ static bool reordering_push(reordering_t* const self, const cy_prio_t priority, const cy_message_ts_t message) { - assert(self->subscriber->params.reordering_window >= 0); - assert(self->topic->cy == self->subscriber->root->cy); + CY_ASSERT(self->subscriber->params.reordering_window >= 0); + CY_ASSERT(self->topic->cy == self->subscriber->root->cy); cy_t* const cy = self->topic->cy; // Dispatch the message according to its tag ordering. @@ -3333,7 +3334,7 @@ static bool reordering_push(reordering_t* const self, (uintmax_t)tag, (uintmax_t)lin_tag, (uintmax_t)self->last_ejected_lin_tag); - assert(self->interned_count == 0); // The above logic will have emptied the interned messages in this case. + CY_ASSERT(self->interned_count == 0); // The above logic will have emptied the interned messages in this case. reordering_resequence(self, tag); lin_tag = tag - self->tag_baseline; } @@ -3343,8 +3344,8 @@ static bool reordering_push(reordering_t* const self, // It may still be a duplicate if somehow it made it past the topic-wise duplicate filter, so we check for that too. // For the assertion to hold, we must ensure that the reordering capacity is at least 4, otherwise the resequencing // logic would set the baseline too low for the assertion to hold. - assert(lin_tag > (self->last_ejected_lin_tag + 1U)); - assert(lin_tag <= (self->last_ejected_lin_tag + capacity)); + CY_ASSERT(lin_tag > (self->last_ejected_lin_tag + 1U)); + CY_ASSERT(lin_tag <= (self->last_ejected_lin_tag + capacity)); reordering_slot_t* const slot = mem_alloc_zero(cy, sizeof(reordering_slot_t)); if (slot == NULL) { CY_TRACE(cy, @@ -3375,11 +3376,11 @@ static bool reordering_push(reordering_t* const self, slot->priority = priority; slot->message = message; self->interned_count++; - assert((self->interned_count == 1) || olga_is_pending(&cy->olga, &self->timeout)); + CY_ASSERT((self->interned_count == 1) || olga_is_pending(&cy->olga, &self->timeout)); // Re-arm against the current head-of-line slot. A newly inserted lower lin_tag may need a later deadline. reordering_slot_t* const first_slot = CAVL2_TO_OWNER(cavl2_min(self->interned_by_lin_tag), reordering_slot_t, index_lin_tag); - assert(first_slot != NULL); + CY_ASSERT(first_slot != NULL); const cy_us_t deadline = first_slot->message.timestamp + self->subscriber->params.reordering_window; olga_defer(&cy->olga, deadline, self, reordering_on_window_expiration, &self->timeout); return true; // Interned messages will eventually be ejected and seen by the application. @@ -3389,7 +3390,7 @@ static void reordering_destroy(reordering_t* const self, const bool silenced) { reordering_eject_all(self, silenced); delist(&self->subscriber->list_reordering_by_recency, &self->list_recency); - assert(cavl2_is_inserted(self->subscriber->index_reordering_by_remote_id, &self->index)); + CY_ASSERT(cavl2_is_inserted(self->subscriber->index_reordering_by_remote_id, &self->index)); cavl2_remove(&self->subscriber->index_reordering_by_remote_id, &self->index); mem_free(self->subscriber->root->cy, self); } @@ -3482,7 +3483,7 @@ static bool on_message(cy_t* const cy, // If we chose to do that, we would need to scan all couplings and subscribers. return false; // The remote will retransmit and we might be able to accept it then. } - assert(dedup->remote_id == lane.id); + CY_ASSERT(dedup->remote_id == lane.id); dedup_touch(dedup, topic, message.timestamp); if (dedup_check(dedup, tag)) { CY_TRACE(cy, "🍒 Dup N%016jx tag=%016jx", (uintmax_t)lane.id, (uintmax_t)tag); @@ -3496,7 +3497,7 @@ static bool on_message(cy_t* const cy, while (cpl != NULL) { const cy_topic_coupling_t* const next_cpl = cpl->next; subscriber_t* sub = cpl->root->head; - assert(sub != NULL); // Otherwise it should have been removed from the coupling list. + CY_ASSERT(sub != NULL); // Otherwise it should have been removed from the coupling list. while (sub != NULL) { subscriber_t* const next_sub = sub->next; if (sub->disposed) { // Skip to avoid acknowledging the message erroneously. @@ -3522,9 +3523,9 @@ static bool on_message(cy_t* const cy, reordering_t, index); if (rr != NULL) { // Simply ignore on OOM, nothing we can do. - assert(rr->remote_id == lane.id); - assert(rr->topic == topic); - assert(rr->subscriber == sub); + CY_ASSERT(rr->remote_id == lane.id); + CY_ASSERT(rr->topic == topic); + CY_ASSERT(rr->subscriber == sub); rr->unicast_ctx = lane.ctx; // keep the latest known return path discovery from the transport if (reordering_push(rr, tag, lane.prio, message)) { // NOTE: If the subscriber is destroyed while there are messages interned in the reordering @@ -3548,8 +3549,9 @@ static bool on_message(cy_t* const cy, } // The entry cannot have been reaped since dedup_touch: the library is non-reentrant. if (reliable && acknowledge) { - assert(dedup != NULL); - assert(dedup->remote_id == lane.id); // Still the entry we touched; the subscriber loop cannot have reaped it. + CY_ASSERT(dedup != NULL); + CY_ASSERT(dedup->remote_id == + lane.id); // Still the entry we touched; the subscriber loop cannot have reaped it. dedup_commit(dedup, tag); } return acknowledge; @@ -3561,10 +3563,10 @@ static size_t subscription_extent_w_overhead(const cy_topic_t* const topic) { size_t total = 0; const cy_topic_coupling_t* cpl = topic->couplings; - assert(cpl != NULL); + CY_ASSERT(cpl != NULL); while (cpl != NULL) { const subscriber_t* sub = cpl->root->head; - assert(sub != NULL); + CY_ASSERT(sub != NULL); while (sub != NULL) { total = larger(total, sub->params.extent_pure); sub = sub->next; @@ -3607,8 +3609,8 @@ static void* wkv_cb_couple_new_subscription(const wkv_event_t evt) // A subscriber root corresponds to a unique subscription name (with possible wildcards), hosting at least 1 subscriber. static cy_err_t ensure_subscriber_root(cy_t* const cy, const cy_resolved_t resolved, subscriber_root_t** const out_root) { - assert((cy != NULL) && (resolved.name.str != NULL) && (resolved.name.len > 0U) && (out_root != NULL)); - assert((resolved.pin == UINT16_MAX) || resolved.verbatim); // enforced during name resolution + CY_ASSERT((cy != NULL) && (resolved.name.str != NULL) && (resolved.name.len > 0U) && (out_root != NULL)); + CY_ASSERT((resolved.pin == UINT16_MAX) || resolved.verbatim); // enforced during name resolution // Find or allocate a tree node. If exists, return as-is. wkv_node_t* const node = wkv_set(&cy->subscribers_by_name, resolved.name); @@ -3619,7 +3621,8 @@ static cy_err_t ensure_subscriber_root(cy_t* const cy, const cy_resolved_t resol subscriber_root_t* const root = (subscriber_root_t*)node->value; *out_root = root; if (root->needs_scouting) { - assert((resolved.pin == UINT16_MAX) && !resolved.verbatim); // can't pin patterns; can only scout patterns + CY_ASSERT((resolved.pin == UINT16_MAX) && + !resolved.verbatim); // can't pin patterns; can only scout patterns const cy_err_t err = do_send_scout(cy, cy_now(cy), resolved.name); root->needs_scouting = err != CY_OK; ON_ASYNC_ERROR_IF(cy, NULL, err); @@ -3646,7 +3649,7 @@ static cy_err_t ensure_subscriber_root(cy_t* const cy, const cy_resolved_t resol mem_free(cy, root); return CY_ERR_MEMORY; } - assert(root->index_pattern->value == NULL); + CY_ASSERT(root->index_pattern->value == NULL); root->index_pattern->value = root; const cy_err_t err = do_send_scout(cy, cy_now(cy), resolved.name); root->needs_scouting = err != CY_OK; @@ -3668,13 +3671,13 @@ static cy_err_t ensure_subscriber_root(cy_t* const cy, const cy_resolved_t resol static subscriber_t* subscribe(cy_t* const cy, const cy_str_t name, const subscriber_params_t params) { - assert((cy != NULL) && (params.reordering_window >= -1)); + CY_ASSERT((cy != NULL) && (params.reordering_window >= -1)); char name_buf[CY_TOPIC_NAME_MAX + 1U]; const cy_resolved_t resolved = cy_resolve(cy, name, sizeof(name_buf), name_buf); if (resolved.name.len > CY_TOPIC_NAME_MAX) { return NULL; } - assert((resolved.pin == UINT16_MAX) || resolved.verbatim); // enforced during name resolution + CY_ASSERT((resolved.pin == UINT16_MAX) || resolved.verbatim); // enforced during name resolution name_buf[resolved.name.len] = 0; // this is not needed for the logic but helps with tracing (if enabled) subscriber_t* const sub = future_new(cy, &subscriber_vtable, sizeof(subscriber_t)); if (sub == NULL) { @@ -3688,7 +3691,7 @@ static subscriber_t* subscribe(cy_t* const cy, const cy_str_t name, const subscr mem_free(cy, sub); return NULL; } - assert(sub->root != NULL); + CY_ASSERT(sub->root != NULL); sub->next = sub->root->head; sub->root->head = sub; if (NULL != wkv_match(&cy->topics_by_name, resolved.name, sub, wkv_cb_couple_new_subscription)) { @@ -3772,8 +3775,8 @@ void cy_subscriber_timeout_set(cy_future_t* const future, const cy_us_t timeout) { if (cy_is_subscriber(future)) { subscriber_t* const self = (subscriber_t*)future; - assert(!self->disposed); // use after free - assert((self->last_arrival.message.timestamp >= 0) && (self->last_arrival.message.timestamp < INT64_MAX)); + CY_ASSERT(!self->disposed); // use after free + CY_ASSERT((self->last_arrival.message.timestamp >= 0) && (self->last_arrival.message.timestamp < INT64_MAX)); self->params.liveness_timeout = sooner(later(0, timeout), KILO * MEGA * MEGA); // Any argument resets the pending liveness error. It may re-appear later. if (self->error == CY_ERR_LIVENESS) { @@ -3805,20 +3808,20 @@ cy_substitution_set_t cy_subscriber_substitutions(const cy_future_t* const futur cy_substitution_set_t out = { .count = 0, .substitutions = NULL }; if (cy_is_subscriber(future)) { const subscriber_t* const self = (const subscriber_t*)future; - assert(!self->disposed); // use after free - if (self->verbatim) { // instant result for verbatim subscribers, no need to scan. + CY_ASSERT(!self->disposed); // use after free + if (self->verbatim) { // instant result for verbatim subscribers, no need to scan. static const cy_substitution_t sentinel; out.substitutions = &sentinel; } else if (topic != NULL) { const cy_topic_coupling_t* cpl = topic->couplings; while (cpl != NULL) { const subscriber_t* sub = cpl->root->head; - assert(sub != NULL); // Otherwise it should have been removed from the coupling list. + CY_ASSERT(sub != NULL); // Otherwise it should have been removed from the coupling list. while (sub != NULL) { if (sub == self) { out.count = cpl->substitution_count; out.substitutions = cpl->substitutions; - assert(out.substitutions != NULL); // never NULL even if empty + CY_ASSERT(out.substitutions != NULL); // never NULL even if empty return out; } sub = sub->next; @@ -3834,7 +3837,7 @@ cy_substitution_set_t cy_subscriber_substitutions(const cy_future_t* const futur static void topic_decouple_subscriber_root(cy_topic_t* const topic, const subscriber_root_t* const root) { - assert((topic != NULL) && (root != NULL)); + CY_ASSERT((topic != NULL) && (root != NULL)); const cy_t* const cy = topic->cy; cy_topic_coupling_t** cpl = &topic->couplings; while (*cpl != NULL) { @@ -3866,8 +3869,8 @@ static void subscriber_destroy(subscriber_t* const self) { cy_t* const cy = self->base.cy; subscriber_root_t* const root = self->root; - assert((root != NULL) && (root->cy == cy)); - assert(self->base.callback == NULL); // Must have been reset beforehand by the future framework. + CY_ASSERT((root != NULL) && (root->cy == cy)); + CY_ASSERT(self->base.callback == NULL); // Must have been reset beforehand by the future framework. future_deadline_disarm(&self->base); #if CY_CONFIG_TRACE char name[CY_TOPIC_NAME_MAX + 1]; @@ -3878,18 +3881,18 @@ static void subscriber_destroy(subscriber_t* const self) // Drop all pending ordered messages first because the states keep pointers into topic couplings. while (self->index_reordering_by_remote_id != NULL) { reordering_t* const rr = CAVL2_TO_OWNER(cavl2_min(self->index_reordering_by_remote_id), reordering_t, index); - assert(rr != NULL); + CY_ASSERT(rr != NULL); reordering_destroy(rr, true); } - assert(self->list_reordering_by_recency.head == NULL); - assert(self->list_reordering_by_recency.tail == NULL); + CY_ASSERT(self->list_reordering_by_recency.head == NULL); + CY_ASSERT(self->list_reordering_by_recency.tail == NULL); // Delist this subscriber from the root. subscriber_t** sub = &root->head; while ((*sub != NULL) && (*sub != self)) { sub = &(*sub)->next; } - assert(*sub == self); + CY_ASSERT(*sub == self); if (*sub == self) { *sub = self->next; // cppcheck-suppress nullPointerRedundantCheck } @@ -3925,10 +3928,10 @@ static cy_err_t do_respond(cy_breadcrumb_t* const breadcrumb, const header_type_t type, const byte_t tag) { - assert((breadcrumb != NULL) && (breadcrumb->cy != NULL) && (deadline >= 0)); + CY_ASSERT((breadcrumb != NULL) && (breadcrumb->cy != NULL) && (deadline >= 0)); // Compose the header. - assert(breadcrumb->seqno < (SEQNO48_MASK - 1U)); // Sanity check; this value is not practically reachable. + CY_ASSERT(breadcrumb->seqno < (SEQNO48_MASK - 1U)); // Sanity check; this value is not practically reachable. byte_t header[HEADER_BYTES] = { (byte_t)type, tag }; (void)serialize_u48(&header[2], breadcrumb->seqno); (void)serialize_u64(&header[8], breadcrumb->topic_hash); @@ -3971,10 +3974,10 @@ static cy_err_t respond_future_error(const cy_future_t* const base) { return ((c static void respond_future_timeout(cy_future_t* const base, const cy_us_t scheduled, const cy_us_t now) { - assert(scheduled <= now); // scheduler invariant + CY_ASSERT(scheduled <= now); // scheduler invariant (void)scheduled; respond_future_t* const self = (respond_future_t*)base; - assert(self->breadcrumb.cy == base->cy); + CY_ASSERT(self->breadcrumb.cy == base->cy); cy_t* const cy = base->cy; // If we are supposed to try more attempts (data not yet destroyed) but we are already near the deadline, @@ -3986,18 +3989,18 @@ static void respond_future_timeout(cy_future_t* const base, const cy_us_t schedu // Check completion. if ((self->data == NULL) || (now >= self->deadline)) { // This is the final poll. future_index_remove(base, &cy->respond_futures_by_tag); - assert(base->vtable->done(base)); // timer not restarted - self->error = CY_ERR_DELIVERY; // no response -- not delivered - future_notify(&self->base); // Invalidates the future. Expect disposal. + CY_ASSERT(base->vtable->done(base)); // timer not restarted + self->error = CY_ERR_DELIVERY; // no response -- not delivered + future_notify(&self->base); // Invalidates the future. Expect disposal. return; } // Compute next deadline and decide if it's going to be the last attempt based on the remaining time. - assert(now < self->deadline); + CY_ASSERT(now < self->deadline); self->ack_timeout *= 2; // exponential backoff const cy_us_t ack_deadline = sooner(self->ack_timeout + now, self->deadline); // manage possible scheduler lag const bool last_attempt = ack_is_last_attempt(ack_deadline, self->ack_timeout, self->deadline); - assert(ack_deadline > now); + CY_ASSERT(ack_deadline > now); // Send the message. const cy_err_t er = do_respond(&self->breadcrumb, ack_deadline, *self->data, header_rsp_rel, self->tag); @@ -4013,15 +4016,15 @@ static void respond_future_timeout(cy_future_t* const base, const cy_us_t schedu self->data = NULL; future_deadline_arm(base, self->deadline); } else { - assert(ack_deadline < self->deadline); + CY_ASSERT(ack_deadline < self->deadline); future_deadline_arm(base, ack_deadline); } // Notify if any errors occurred, but we are not done yet. if ((er != CY_OK) || sched_lag_error) { - assert(self->error != CY_OK); - assert(!base->vtable->done(base)); // Not done yet -- timer pending. - future_notify(&self->base); // Invalidates the future. Expect disposal. + CY_ASSERT(self->error != CY_OK); + CY_ASSERT(!base->vtable->done(base)); // Not done yet -- timer pending. + future_notify(&self->base); // Invalidates the future. Expect disposal. } } @@ -4044,7 +4047,7 @@ static const cy_future_vtable_t respond_future_vtable = { .done = respond_fut static void respond_future_on_ack(respond_future_t* const self, const bool positive_ack) { cy_t* const cy = self->base.cy; - assert(!self->base.vtable->done(&self->base)); + CY_ASSERT(!self->base.vtable->done(&self->base)); self->error = positive_ack ? CY_OK : CY_ERR_NACK; // Overwrite previous error -- assume it has been seen. future_deadline_disarm(&self->base); future_index_remove(&self->base, &cy->respond_futures_by_tag); @@ -4059,7 +4062,7 @@ static uint64_t respond_key(const uint64_t remote_id, const uint64_t seqno, const byte_t tag) { - assert(seqno <= SEQNO48_MASK); + CY_ASSERT(seqno <= SEQNO48_MASK); // This simple and fast hash should suffice. We could use rapidhash but it's likely an overkill. // Message tag and seqno change their LSb quickly, which is why we shift seqno to the left (it's only 48 bits wide). // The tag is shifted left for the same reason -- we want it to reside in the area where bits are mostly static. @@ -4149,10 +4152,10 @@ cy_future_t* cy_respond_reliable(cy_breadcrumb_t* const breadcrumb, const cy_us_ static void topic_destroy(cy_topic_t* const topic) { - assert((topic != NULL) && (topic->cy != NULL)); - assert(topic->pub_count == 0); - assert(topic->pub_futures_by_tag == NULL); - assert(topic->couplings == NULL); // removed when unsubscribed + CY_ASSERT((topic != NULL) && (topic->cy != NULL)); + CY_ASSERT(topic->pub_count == 0); + CY_ASSERT(topic->pub_futures_by_tag == NULL); + CY_ASSERT(topic->couplings == NULL); // removed when unsubscribed cy_t* const cy = topic->cy; CY_TRACE(cy, "🗑️ %s", topic_repr(topic).str); @@ -4173,20 +4176,20 @@ static void topic_destroy(cy_topic_t* const topic) // Remove subscriber associations. while (topic->assoc_by_remote_id != NULL) { association_t* const ass = CAVL2_TO_OWNER(cavl2_min(topic->assoc_by_remote_id), association_t, index_remote_id); - assert(ass != NULL); - assert(ass->pending_count == 0); + CY_ASSERT(ass != NULL); + CY_ASSERT(ass->pending_count == 0); association_forget(topic, ass); } - assert(topic->assoc_count == 0); + CY_ASSERT(topic->assoc_count == 0); // Remove message deduplication states. while (topic->sub_index_dedup_by_remote_id != NULL) { dedup_t* const dd = CAVL2_TO_OWNER(cavl2_min(topic->sub_index_dedup_by_remote_id), dedup_t, index_remote_id); - assert(dd != NULL); + CY_ASSERT(dd != NULL); dedup_destroy(dd, topic); } - assert(topic->sub_list_dedup_by_recency.head == NULL); - assert(topic->sub_list_dedup_by_recency.tail == NULL); + CY_ASSERT(topic->sub_list_dedup_by_recency.head == NULL); + CY_ASSERT(topic->sub_list_dedup_by_recency.tail == NULL); // Remove any zombie request futures that may be left behind to manage retransmissions. // This is lifetime-safe because the API contract requires that the application must destroy pending futures @@ -4195,7 +4198,7 @@ static void topic_destroy(cy_topic_t* const topic) // expires the zombie request futures are likely going to be destroyed on timeout anyway. while (topic->request_futures_by_tag != NULL) { request_future_t* const future = (request_future_t*)topic->request_futures_by_tag; - assert(future->finalized); // Otherwise, the application forgot to destroy the future! + CY_ASSERT(future->finalized); // Otherwise, the application forgot to destroy the future! request_future_destroy(future); } @@ -4216,7 +4219,7 @@ static void topic_destroy(cy_topic_t* const topic) delist(&cy->list_implicit, &topic->list_implicit); // cavl2_remove_if(&cy->topics_by_subject_id, &topic->index_subject_id); - assert(cavl2_is_inserted(cy->topics_by_hash, &topic->index_hash)); + CY_ASSERT(cavl2_is_inserted(cy->topics_by_hash, &topic->index_hash)); cavl2_remove(&cy->topics_by_hash, &topic->index_hash); // if (topic->index_name != NULL) { @@ -4262,11 +4265,11 @@ static void destroy_disposed_subscribers(cy_t* const cy) { while (!wkv_is_empty(&cy->subscribers_by_name)) { const wkv_node_t* const node = wkv_at(&cy->subscribers_by_name, 0); - assert((node != NULL) && (node->value != NULL)); + CY_ASSERT((node != NULL) && (node->value != NULL)); subscriber_root_t* const root = (subscriber_root_t*)node->value; - assert((root != NULL) && (root->head != NULL)); + CY_ASSERT((root != NULL) && (root->head != NULL)); subscriber_t* const sub = root->head; - assert(sub->disposed); + CY_ASSERT(sub->disposed); if (!sub->disposed) { break; } @@ -4390,8 +4393,8 @@ cy_t* cy_new(cy_platform_t* const platform, const cy_str_t home, const cy_str_t memset(cy, 0, sizeof(cy_t) + home_len + ns_len + 2); // Zero the entire allocation incl. NUL terminators. cy->home = name_normalize(home, home_len, (char*)cy + sizeof(cy_t)); cy->ns = name_normalize(effective_ns, ns_len, (char*)cy + sizeof(cy_t) + home_len + 1); - assert((cy->home.str != NULL) && (cy->home.len == home_len) && (cy->home.str[cy->home.len] == '\0')); - assert((cy->ns.str != NULL) && (cy->ns.len == ns_len) && (cy->ns.str[cy->ns.len] == '\0')); + CY_ASSERT((cy->home.str != NULL) && (cy->home.len == home_len) && (cy->home.str[cy->home.len] == '\0')); + CY_ASSERT((cy->ns.str != NULL) && (cy->ns.len == ns_len) && (cy->ns.str[cy->ns.len] == '\0')); cy->platform = platform; platform->cy = cy; @@ -4444,7 +4447,7 @@ cy_t* cy_new(cy_platform_t* const platform, const cy_str_t home, const cy_str_t const uint32_t broadcast_subject_id = (uint32_t)((1ULL << (byte_t)(log2_floor(CY_SUBJECT_ID_MAX(platform->subject_id_modulus)) + 1)) - 1U); cy->gossip_shard_count = broadcast_subject_id - (CY_SUBJECT_ID_MAX(platform->subject_id_modulus) + 1U); - assert((cy->gossip_shard_count > 0) && (cy->gossip_shard_count < platform->subject_id_modulus)); // sanity + CY_ASSERT((cy->gossip_shard_count > 0) && (cy->gossip_shard_count < platform->subject_id_modulus)); // sanity // Set up the broadcast subject readers/writers. cy->broad_reader = @@ -4483,9 +4486,9 @@ void cy_destroy(cy_t* const cy) // Ensure the user has cleaned up beforehand. // We are unable to destroy user-owner objects like publishers/subscribers/futures because we don't own them. - assert(wkv_is_empty(&cy->subscribers_by_name)); - assert(wkv_is_empty(&cy->subscribers_by_pattern)); - assert(cy->respond_futures_by_tag == NULL); // All pending response futures must be destroyed. + CY_ASSERT(wkv_is_empty(&cy->subscribers_by_name)); + CY_ASSERT(wkv_is_empty(&cy->subscribers_by_pattern)); + CY_ASSERT(cy->respond_futures_by_tag == NULL); // All pending response futures must be destroyed. // Remove global subject reader & writer. if (cy->broad_reader != NULL) { @@ -4507,21 +4510,21 @@ void cy_destroy(cy_t* const cy) // There may still be implicit topics left, but they must have no user-owned entities attached anymore. while (cy->topics_by_hash != NULL) { cy_topic_t* const topic = cy_topic_iter_first(cy); - assert(topic != NULL); - assert(topic->pub_futures_by_tag == NULL); // Caller must destroy futures. - assert(topic->pub_count == 0); // Caller must destroy publishers. - assert(topic->couplings == NULL); // Caller must destroy subscribers. - assert(is_implicit(topic)); + CY_ASSERT(topic != NULL); + CY_ASSERT(topic->pub_futures_by_tag == NULL); // Caller must destroy futures. + CY_ASSERT(topic->pub_count == 0); // Caller must destroy publishers. + CY_ASSERT(topic->couplings == NULL); // Caller must destroy subscribers. + CY_ASSERT(is_implicit(topic)); topic_destroy(topic); } - assert(wkv_is_empty(&cy->topics_by_name)); - assert(cy->writers == NULL); // All writer registry entries released by topic_destroy. - assert(cy->readers == NULL); // All reader registry entries released by topic_destroy. + CY_ASSERT(wkv_is_empty(&cy->topics_by_name)); + CY_ASSERT(cy->writers == NULL); // All writer registry entries released by topic_destroy. + CY_ASSERT(cy->readers == NULL); // All reader registry entries released by topic_destroy. // Drain the remap table: the to-string values are owned by this cy_t and must be freed here. while (!wkv_is_empty(&cy->remap)) { wkv_node_t* const n = wkv_at(&cy->remap, 0); - assert((n != NULL) && (n->value != NULL)); + CY_ASSERT((n != NULL) && (n->value != NULL)); mem_free(cy, n->value); wkv_del(&cy->remap, n); } @@ -4581,7 +4584,7 @@ cy_err_t cy_spin_until(cy_t* const cy, const cy_us_t deadline) cy_us_t cy_now(const cy_t* const cy) { const cy_us_t out = cy->platform->vtable->now(cy->platform); - assert(out >= 0); + CY_ASSERT(out >= 0); return out; } @@ -4672,15 +4675,15 @@ cy_topic_t* cy_topic_find_by_name(const cy_t* const cy, const cy_str_t name) { const wkv_node_t* const node = wkv_get(&cy->topics_by_name, name); cy_topic_t* const topic = (node != NULL) ? (cy_topic_t*)node->value : NULL; - assert(topic == cy_topic_find_by_hash(cy, rapidhash(name.str, name.len))); + CY_ASSERT(topic == cy_topic_find_by_hash(cy, rapidhash(name.str, name.len))); return topic; } cy_topic_t* cy_topic_find_by_hash(const cy_t* const cy, const uint64_t hash) { - assert(cy != NULL); + CY_ASSERT(cy != NULL); cy_topic_t* const topic = (cy_topic_t*)cavl2_find(cy->topics_by_hash, &hash, &cavl_comp_topic_hash); - assert((topic == NULL) || (topic->hash == hash)); + CY_ASSERT((topic == NULL) || (topic->hash == hash)); return topic; } @@ -4713,7 +4716,7 @@ static void on_message_ack(cy_t* const cy, const bool positive, const cy_lane_t lane) { - assert(topic != NULL); + CY_ASSERT(topic != NULL); // Protect against acks that are clearly invalid, so that we don't blow up the association set unnecessarily. // The max lag limits the oldest ack we can accept; it is chosen to be large enough to fit any valid use case. @@ -4750,7 +4753,7 @@ static void on_message_ack(cy_t* const cy, // Update the state of the local subscriber association. // NACK for an association that is not currently used by any publisher future allows immediate removal. - assert(topic->assoc_count > 0); + CY_ASSERT(topic->assoc_count > 0); ass->last_seen = ts; ass->unicast_ctx = lane.ctx; // Always update the latest return path discovery state. if (seqno >= ass->seqno_witness) { // Prevent delayed acks from overwriting newer states. @@ -4764,7 +4767,7 @@ static void on_message_ack(cy_t* const cy, } // There are futures that might be interested. - assert(positive || (ass->pending_count > 0) || (seqno < ass->seqno_witness)); + CY_ASSERT(positive || (ass->pending_count > 0) || (seqno < ass->seqno_witness)); publish_future_t* const future = (publish_future_t*)future_index_lookup(topic->pub_futures_by_tag, tag); if (future != NULL) { publish_future_on_ack(future, lane.id, positive); @@ -4805,7 +4808,7 @@ static void send_response_ack(cy_t* const cy, const bool positive, const cy_us_t deadline) { - assert(seqno <= SEQNO48_MASK); + CY_ASSERT(seqno <= SEQNO48_MASK); byte_t header[HEADER_BYTES] = { (byte_t)(positive ? header_rsp_ack : header_rsp_nack), tag }; (void)serialize_u48(&header[2], seqno); (void)serialize_u64(&header[8], hash); @@ -4830,8 +4833,8 @@ void cy_on_message(cy_platform_t* const platform, const cy_message_ts_t message) { cy_t* const cy = platform->cy; - assert((cy != NULL) && (message.timestamp >= 0)); - assert(message.content->refcount == 1); + CY_ASSERT((cy != NULL) && (message.timestamp >= 0)); + CY_ASSERT(message.content->refcount == 1); byte_t header[HEADER_BYTES] = { 0 }; if (cy_message_read(message.content, 0, HEADER_BYTES, header) != HEADER_BYTES) { goto bad_message; @@ -4867,8 +4870,8 @@ void cy_on_message(cy_platform_t* const platform, // Process the message if the topic is known. bool accepted = false; if (topic != NULL) { - assert((topic->sub_reader == NULL) || - (topic_subject_id(topic) == topic->sub_reader->handle->subject_id)); + CY_ASSERT((topic->sub_reader == NULL) || + (topic_subject_id(topic) == topic->sub_reader->handle->subject_id)); // We have the topic, which may or may not be using the same subject-ID. If we use this subject-ID // for another topic, then it constitutes both a divergence and a collision. The correct handling // is to address the divergence by either moving the local topic if the gossiped state is newer, @@ -5065,7 +5068,7 @@ static bool str_valid(const cy_str_t str) { return (str.str != NULL) || (str.len static cy_str_t str_skip(const cy_str_t s, const size_t k) { - assert(str_valid(s) && (k <= s.len)); + CY_ASSERT(str_valid(s) && (k <= s.len)); return (cy_str_t){ .len = s.len - k, .str = &s.str[k] }; } @@ -5074,7 +5077,7 @@ static cy_str_t str_skip(const cy_str_t s, const size_t k) // Example: `foo#123` => `foo`, out_pin=123; `foo#0` => `foo`, out_pin=0; `foo#01` => unchanged (leading zero). static cy_str_t name_consume_pin_suffix(const cy_str_t name, uint16_t* const out_pin) { - assert((out_pin != NULL) && str_valid(name)); + CY_ASSERT((out_pin != NULL) && str_valid(name)); *out_pin = UINT16_MAX; // Scan right-to-left: find '#' preceded only by decimal digits. size_t hash_pos = name.len; @@ -5129,7 +5132,7 @@ static bool name_is_absolute(const cy_str_t n) { return (n.str != NULL) && (n.le // Returns the length of the normalized string, or SIZE_MAX if the input contains invalid characters. static size_t name_normalized_len(const cy_str_t name) { - assert(str_valid(name)); + CY_ASSERT(str_valid(name)); size_t out_len = 0U; bool pending_sep = false; for (size_t i = 0; i < name.len; i++) { @@ -5152,12 +5155,12 @@ static size_t name_normalized_len(const cy_str_t name) static void name_copy_normalized_forward(const cy_str_t name, char* const dest) { - assert(str_valid(name) && (dest != NULL)); + CY_ASSERT(str_valid(name) && (dest != NULL)); char* out = dest; bool pending_sep = false; for (size_t i = 0; i < name.len; i++) { const char c = name.str[i]; - assert(is_valid_char(c)); + CY_ASSERT(is_valid_char(c)); if (c == cy_name_sep) { pending_sep = out > dest; // skip duplicate and leading separators continue; @@ -5173,7 +5176,7 @@ static void name_copy_normalized_forward(const cy_str_t name, char* const dest) // Exact in-place normalization is supported; arbitrary partial overlap is not guaranteed. static cy_str_t name_normalize(const cy_str_t part, const size_t dest_size, char* const dest) { - assert(((part.str != NULL) || (part.len == 0U)) && (dest != NULL)); + CY_ASSERT(((part.str != NULL) || (part.len == 0U)) && (dest != NULL)); const size_t len = name_normalized_len(part); if ((len == SIZE_MAX) || (len > dest_size)) { return str_invalid; @@ -5221,7 +5224,7 @@ static cy_str_t name_resolve_construct(const cy_str_t name, const size_t dest_size, char* dest) { - assert(str_valid(name) && str_valid(name_space) && str_valid(home) && (dest != NULL)); + CY_ASSERT(str_valid(name) && str_valid(name_space) && str_valid(home) && (dest != NULL)); if (name_is_absolute(name)) { return name_normalize(name, dest_size, dest); } @@ -5298,7 +5301,7 @@ cy_resolved_t cy_name_resolve(const wkv_t* const remap, #define DIAG_FOREACH(cy, diag_call, ...) \ do { \ - assert((cy) != NULL); \ + CY_ASSERT((cy) != NULL); \ for (cy_diag_t* diag = (cy)->diags; diag != NULL;) { \ cy_diag_t* const next = diag->next; \ if ((diag->vtable != NULL) && (diag->vtable->diag_call != NULL)) { \ diff --git a/cy/cy.h b/cy/cy.h index 1bd080d..5cc8015 100644 --- a/cy/cy.h +++ b/cy/cy.h @@ -13,7 +13,22 @@ #pragma once +// Define this macro to include build configuration header. +// Usage example with CMake: "-DCY_CONFIG_HEADER=\"${CMAKE_CURRENT_SOURCE_DIR}/my_cy_config.h\"" +#ifdef CY_CONFIG_HEADER +#include CY_CONFIG_HEADER +#endif + +// By default, this macro resolves to the standard assert(). +// To disable assertion checks completely, make it expand into `(void)(0)`. +#ifndef CY_ASSERT +#include +#define CY_ASSERT(x) assert(x) +#endif + +#define WKV_ASSERT(x) CY_ASSERT(x) #include + #include #include #include diff --git a/cy_can/cy_can.c b/cy_can/cy_can.c index 7323b8d..a6b980b 100644 --- a/cy_can/cy_can.c +++ b/cy_can/cy_can.c @@ -313,7 +313,7 @@ static bool v_canard_tx(canard_t* const self, cy_can_t* const owner = (cy_can_t*)self->user_context; const uint_least8_t len = (uint_least8_t)can_data.size; (void)user_context; - assert(iface_index < owner->iface_count); + CY_ASSERT(iface_index < owner->iface_count); if (fd && (owner->vtable->tx_fd != NULL)) { return owner->vtable->tx_fd(owner->user, deadline, iface_index, extended_can_id, can_data.data, len); } @@ -323,7 +323,7 @@ static bool v_canard_tx(canard_t* const self, static bool v_canard_filter(canard_t* const self, const size_t filter_count, const canard_filter_t* const filters) { cy_can_t* const owner = (cy_can_t*)self->user_context; - assert((owner != NULL) && (owner->vtable != NULL) && (owner->vtable->filter != NULL)); + CY_ASSERT((owner != NULL) && (owner->vtable != NULL) && (owner->vtable->filter != NULL)); return owner->vtable->filter(owner->user, filter_count, filters); } @@ -353,11 +353,11 @@ static void v_on_msg_16b(canard_subscription_t* const self, { (void)transfer_id; subject_reader_t* const reader = (subject_reader_t*)self->user_context; - assert(reader != NULL); + CY_ASSERT(reader != NULL); cy_can_t* const owner = reader->owner; - assert(owner != NULL); + CY_ASSERT(owner != NULL); pending_v1_t* const pending = owner->pending_v1; - assert((pending != NULL) && (pending->message.content == NULL)); + CY_ASSERT((pending != NULL) && (pending->message.content == NULL)); const bool multiframe = (payload.origin.data != NULL); can_message_t* const msg = make_message(owner, multiframe ? 0 : payload.view.size); @@ -397,12 +397,12 @@ static void v_on_msg_13b(canard_subscription_t* const self, { (void)transfer_id; subject_reader_t* const reader = (subject_reader_t*)self->user_context; - assert(reader != NULL); + CY_ASSERT(reader != NULL); cy_can_t* const owner = reader->owner; - assert(owner != NULL); + CY_ASSERT(owner != NULL); pending_v1_t* const pending = owner->pending_v1; subject_reader_pinned_t* const pinned = as_pinned(reader); - assert((pinned != NULL) && (pending != NULL) && (pending->message.content == NULL)); + CY_ASSERT((pinned != NULL) && (pending != NULL) && (pending->message.content == NULL)); const bool multiframe = (payload.origin.data != NULL); const size_t inline_size = HEADER_BYTES + (multiframe ? 0 : payload.view.size); @@ -451,9 +451,9 @@ static void v_on_msg_unicast(canard_subscription_t* const self, { (void)transfer_id; cy_can_t* const owner = (cy_can_t*)self->user_context; - assert(owner != NULL); + CY_ASSERT(owner != NULL); pending_v1_t* const pending = owner->pending_v1; - assert((pending != NULL) && (pending->message.content == NULL)); + CY_ASSERT((pending != NULL) && (pending->message.content == NULL)); const bool multiframe = (payload.origin.data != NULL); can_message_t* const msg = make_message(owner, multiframe ? 0 : payload.view.size); @@ -530,7 +530,7 @@ static cy_err_t v_subject_writer_send(cy_platform_t* const platform, const uint32_t sid = base->subject_id; const uint_least8_t ibm = (uint_least8_t)((1U << owner->iface_count) - 1U); - assert((message.data != NULL) && (message.size >= HEADER_BYTES)); + CY_ASSERT((message.data != NULL) && (message.size >= HEADER_BYTES)); const bool pinned = (sid <= CY_SUBJECT_ID_PINNED_MAX); const bool best_effort = (((const uint_least8_t*)message.data)[0] == 0); // header_msg_be const bool use_13b = pinned && best_effort && topic_is_compat_named(sid, message.data); @@ -599,7 +599,7 @@ static void reader_set_extent(subject_reader_t* const self, const size_t extent) static void tombstone_remove(cy_can_t* const owner, subject_reader_t* const self) { - assert((owner != NULL) && (self != NULL)); + CY_ASSERT((owner != NULL) && (self != NULL)); if (self->prev_tombstone != NULL) { self->prev_tombstone->next_tombstone = self->next_tombstone; } else { @@ -616,7 +616,7 @@ static void tombstone_remove(cy_can_t* const owner, subject_reader_t* const self static void tombstone_enqueue(cy_can_t* const owner, subject_reader_t* const self) { - assert((owner != NULL) && (self != NULL)); + CY_ASSERT((owner != NULL) && (self != NULL)); self->prev_tombstone = owner->tombstone_tail; self->next_tombstone = NULL; if (owner->tombstone_tail != NULL) { @@ -644,11 +644,11 @@ static subject_reader_t* reader_try_revive(cy_can_t* const owner, const uint32_t return NULL; } subject_reader_t* const self = (subject_reader_t*)incumbent->user_context; - assert((self != NULL) && (self->owner == owner) && (self->base.subject_id == subject_id)); + CY_ASSERT((self != NULL) && (self->owner == owner) && (self->base.subject_id == subject_id)); if (as_pinned(self) != NULL) { canard_subscription_t* const incumbent_13b = canard_find_subscription(&owner->canard, canard_kind_message_13b, (uint16_t)subject_id); - assert((incumbent_13b != NULL) && (((subject_reader_t*)incumbent_13b->user_context) == self)); + CY_ASSERT((incumbent_13b != NULL) && (((subject_reader_t*)incumbent_13b->user_context) == self)); if ((incumbent_13b == NULL) || (((subject_reader_t*)incumbent_13b->user_context) != self)) { return NULL; } @@ -663,7 +663,7 @@ static subject_reader_t* reader_try_revive(cy_can_t* const owner, const uint32_t /// Finalize a reader: unsubscribe from canard and free memory. Does NOT unlink from any list. static void reader_finalize(cy_can_t* const owner, subject_reader_t* const self) { - assert((owner != NULL) && (self != NULL)); + CY_ASSERT((owner != NULL) && (self != NULL)); if (owner->base.cy != NULL) { CY_TRACE(owner->base.cy, "S%08jx ptr=%p", (uintmax_t)self->base.subject_id, (void*)self); } @@ -704,7 +704,7 @@ static cy_subject_reader_t* v_subject_reader_new(cy_platform_t* const base, extent, CANARD_DEFAULT_TRANSFER_ID_TIMEOUT_us, &sub_vtable_16b); - assert(sub_16b == &self->sub_16b); + CY_ASSERT(sub_16b == &self->sub_16b); if (sub_16b != &self->sub_16b) { owner->vtable->realloc(owner->user, self, 0); return NULL; @@ -715,14 +715,14 @@ static cy_subject_reader_t* v_subject_reader_new(cy_platform_t* const base, subject_reader_pinned_t* const p = (subject_reader_pinned_t*)self; // 13-bit payload does not include the Cy header; we prepend it ourselves. const size_t extent_13b = (extent > HEADER_BYTES) ? (extent - HEADER_BYTES) : 0; - assert(canard_find_subscription(&owner->canard, canard_kind_message_13b, (uint16_t)subject_id) == NULL); + CY_ASSERT(canard_find_subscription(&owner->canard, canard_kind_message_13b, (uint16_t)subject_id) == NULL); canard_subscription_t* const sub_13b = canard_subscribe_13b(&owner->canard, &p->sub_13b, (uint16_t)subject_id, extent_13b, CANARD_DEFAULT_TRANSFER_ID_TIMEOUT_us, &sub_vtable_13b); - assert(sub_13b == &p->sub_13b); + CY_ASSERT(sub_13b == &p->sub_13b); if (sub_13b != &p->sub_13b) { canard_unsubscribe(&owner->canard, &self->sub_16b); owner->vtable->realloc(owner->user, self, 0); @@ -818,7 +818,7 @@ static void ingest_frame(cy_can_t* const owner, const cy_can_rx_t* const frame) const canard_bytes_t can_data = { .size = frame->len, .data = frame->data }; pending_v1_t pending_v1 = { 0 }; pending_v0_t pending_v0 = { 0 }; - assert((owner->pending_v1 == NULL) && (owner->pending_v0 == NULL)); + CY_ASSERT((owner->pending_v1 == NULL) && (owner->pending_v0 == NULL)); owner->pending_v1 = &pending_v1; owner->pending_v0 = &pending_v0; (void)canard_ingest_frame(&owner->canard, frame->timestamp, frame->iface_index, frame->can_id, can_data); @@ -1002,10 +1002,10 @@ void cy_can_destroy(cy_platform_t* const base) if (owner == NULL) { return; } - assert((owner->pending_v1 == NULL) && (owner->pending_v0 == NULL)); + CY_ASSERT((owner->pending_v1 == NULL) && (owner->pending_v0 == NULL)); while (owner->tombstone_head != NULL) { subject_reader_t* const rd = tombstone_pop(owner); - assert(rd != NULL); + CY_ASSERT(rd != NULL); reader_finalize(owner, rd); } canard_unsubscribe(&owner->canard, &owner->unicast_sub); @@ -1033,10 +1033,10 @@ static void v_on_msg_v0(canard_subscription_t* const self, const canard_payload_t payload) { cy_can_v0_subscription_t* const sub = (cy_can_v0_subscription_t*)self->user_context; - assert((sub != NULL) && (sub->owner != NULL)); + CY_ASSERT((sub != NULL) && (sub->owner != NULL)); cy_can_t* const owner = sub->owner; pending_v0_t* const pending = owner->pending_v0; - assert((pending != NULL) && (pending->payload.view.data == NULL)); + CY_ASSERT((pending != NULL) && (pending->payload.view.data == NULL)); pending->subscription = sub; pending->timestamp = timestamp; pending->priority = priority; @@ -1044,7 +1044,7 @@ static void v_on_msg_v0(canard_subscription_t* const self, pending->transfer_id = transfer_id; pending->payload = payload; if (payload.origin.data == NULL) { - assert(payload.view.size <= sizeof(pending->single_frame_data)); + CY_ASSERT(payload.view.size <= sizeof(pending->single_frame_data)); (void)memcpy(pending->single_frame_data, payload.view.data, payload.view.size); pending->payload.view.data = pending->single_frame_data; } diff --git a/cy_can/cy_can_socketcan.c b/cy_can/cy_can_socketcan.c index 6a6bc56..2e77c07 100644 --- a/cy_can/cy_can_socketcan.c +++ b/cy_can/cy_can_socketcan.c @@ -21,7 +21,6 @@ #define RAPIDHASH_COMPACT #include -#include #include #include #include @@ -63,7 +62,7 @@ static bool v_tx_classic(void* const user, { const socketcan_t* const self = (const socketcan_t*)user; (void)deadline; - assert(iface_index < self->iface_count); + CY_ASSERT(iface_index < self->iface_count); struct can_frame frame = { .can_id = can_id | CAN_EFF_FLAG, .can_dlc = len }; if ((data != NULL) && (len > 0)) { (void)memcpy(frame.data, data, (len <= 8) ? len : 8); @@ -81,7 +80,7 @@ static bool v_tx_fd(void* const user, { const socketcan_t* const self = (const socketcan_t*)user; (void)deadline; - assert(iface_index < self->iface_count); + CY_ASSERT(iface_index < self->iface_count); struct canfd_frame frame = { .can_id = can_id | CAN_EFF_FLAG, .len = len, .flags = CANFD_FDF }; if ((data != NULL) && (len > 0)) { (void)memcpy(frame.data, data, (len <= 64) ? len : 64); @@ -214,8 +213,8 @@ static const cy_can_vtable_t socketcan_vtable_classic = { .tx_classic = v_tx_cla static bool socketcan_iface_is_fd_capable(const int sock, const char* const iface_name) { - assert(sock >= 0); - assert(iface_name != NULL); + CY_ASSERT(sock >= 0); + CY_ASSERT(iface_name != NULL); struct ifreq ifr; (void)memset(&ifr, 0, sizeof(ifr)); (void)strncpy(ifr.ifr_name, iface_name, sizeof(ifr.ifr_name) - 1); @@ -227,7 +226,7 @@ static bool socketcan_iface_is_fd_capable(const int sock, const char* const ifac static bool socketcan_set_fd_frames(const socketcan_t* const self, const bool enable) { - assert(self != NULL); + CY_ASSERT(self != NULL); const int value = enable ? 1 : 0; for (uint_least8_t i = 0; i < self->iface_count; i++) { if (setsockopt(self->sock_fd[i], SOL_CAN_RAW, CAN_RAW_FD_FRAMES, &value, sizeof(value)) < 0) { diff --git a/cy_udp_posix/cy_udp_posix.c b/cy_udp_posix/cy_udp_posix.c index 9b9d4e8..bdd1105 100644 --- a/cy_udp_posix/cy_udp_posix.c +++ b/cy_udp_posix/cy_udp_posix.c @@ -114,9 +114,9 @@ static void mem_free(void* const user, const size_t size, void* const pointer) cy_udp_posix_t* const self = (cy_udp_posix_t*)user; (void)size; if (pointer != NULL) { - assert(self->stats.mem.allocated_fragments > 0); + CY_ASSERT(self->stats.mem.allocated_fragments > 0); self->stats.mem.allocated_fragments--; - assert(self->stats.mem.allocated_bytes >= size); + CY_ASSERT(self->stats.mem.allocated_bytes >= size); self->stats.mem.allocated_bytes -= size; memset(pointer, 0xA5, size); // a simple diagnostic aid free(pointer); @@ -198,7 +198,7 @@ static size_t v_message_read(const cy_message_t* const base, const size_t offset static size_t v_message_read_1(const cy_message_t* const base, const size_t offset, const size_t size, void* const dest) { message_t* const self = (message_t*)base; - assert((self->fragment->index_offset.lr[0] == NULL) && (self->fragment->index_offset.lr[1] == NULL)); + CY_ASSERT((self->fragment->index_offset.lr[0] == NULL) && (self->fragment->index_offset.lr[1] == NULL)); size_t out = 0; if (offset < self->size) { out = smaller(size, self->size - offset); @@ -250,7 +250,7 @@ struct subject_writer_t static cy_subject_writer_t* v_subject_writer_new(cy_platform_t* const base, const uint32_t subject_id) { (void)subject_id; - assert(subject_id <= UDPARD_IPv4_SUBJECT_ID_MAX); + CY_ASSERT(subject_id <= UDPARD_IPv4_SUBJECT_ID_MAX); cy_udp_posix_t* const owner = (cy_udp_posix_t*)base; subject_writer_t* const self = mem_alloc_zero(owner, sizeof(subject_writer_t)); if (self != NULL) { @@ -269,7 +269,7 @@ static void v_subject_writer_destroy(cy_platform_t* const platform, cy_subject_w { cy_udp_posix_t* const owner = (cy_udp_posix_t*)platform; subject_writer_t* const self = (subject_writer_t*)base; - assert(owner->stats.subject_writer_count > 0); + CY_ASSERT(owner->stats.subject_writer_count > 0); owner->stats.subject_writer_count--; CY_TRACE(owner->base.cy, "🔇 n_writers=%zu ptr=%p", owner->stats.subject_writer_count, (void*)self); mem_free(owner, sizeof(subject_writer_t), self); @@ -381,7 +381,7 @@ static cy_subject_reader_t* v_subject_reader_new(cy_platform_t* const base, const uint32_t subject_id, const size_t extent) { - assert(subject_id <= UDPARD_IPv4_SUBJECT_ID_MAX); + CY_ASSERT(subject_id <= UDPARD_IPv4_SUBJECT_ID_MAX); cy_udp_posix_t* const owner = (cy_udp_posix_t*)base; subject_reader_t* self = mem_alloc_zero(owner, sizeof(subject_reader_t)); if (self != NULL) { @@ -440,7 +440,7 @@ static cy_subject_reader_t* v_subject_reader_new(cy_platform_t* const base, owner->reader_tail = self; } owner->reader_head = self; - assert((owner->reader_head != NULL) == (owner->reader_tail != NULL)); + CY_ASSERT((owner->reader_head != NULL) == (owner->reader_tail != NULL)); } reject: CY_TRACE(owner->base.cy, @@ -453,8 +453,8 @@ static cy_subject_reader_t* v_subject_reader_new(cy_platform_t* const base, static void subject_reader_destroy(cy_udp_posix_t* const owner, subject_reader_t* const self) { - assert(self->port.user == self); - assert(self->tombstone); + CY_ASSERT(self->port.user == self); + CY_ASSERT(self->tombstone); // Delist. if (self->prev != NULL) { @@ -471,7 +471,7 @@ static void subject_reader_destroy(cy_udp_posix_t* const owner, subject_reader_t } self->prev = NULL; self->next = NULL; - assert((owner->reader_head != NULL) == (owner->reader_tail != NULL)); + CY_ASSERT((owner->reader_head != NULL) == (owner->reader_tail != NULL)); // Cleanup the libudpard port and the sockets. udpard_rx_port_free(&owner->udpard_rx, &self->port); @@ -480,7 +480,7 @@ static void subject_reader_destroy(cy_udp_posix_t* const owner, subject_reader_t } // Free the memory and update the stats. - assert(owner->stats.subject_reader_count > 0); + CY_ASSERT(owner->stats.subject_reader_count > 0); owner->stats.subject_reader_count--; CY_TRACE(owner->base.cy, "🔕 n_readers=%zu ptr=%p", owner->stats.subject_reader_count, (void*)self); mem_free(owner, sizeof(subject_reader_t), self); @@ -490,8 +490,8 @@ static void v_subject_reader_tombstone(cy_platform_t* const platform, cy_subject { cy_udp_posix_t* const owner = (cy_udp_posix_t*)platform; subject_reader_t* const self = (subject_reader_t*)base; - assert(self->port.user == self); - assert(!self->tombstone); + CY_ASSERT(self->port.user == self); + CY_ASSERT(!self->tombstone); self->tombstone = true; // Close sockets now to stop further reads while we defer the final teardown. // This also makes same-subject recreation safe: unlike libcanard, libudpard has no global uniqueness rule for @@ -575,13 +575,13 @@ static void read_socket(cy_udp_posix_t* const self, udp_wrapper_t* const sock, const uint_fast8_t iface_index) { - assert((self->iface_bitmap & (1U << iface_index)) != 0); - assert(iface_index <= CY_UDP_POSIX_IFACE_COUNT_MAX); - assert(is_valid_ip(self->local_ip[iface_index])); - assert(udp_wrapper_is_open(sock)); - assert((self->stats.subject_reader_count == 0) == (self->reader_head == NULL)); - assert((self->stats.subject_reader_count == 0) == (self->reader_tail == NULL)); - assert((reader == NULL) || !reader->tombstone); + CY_ASSERT((self->iface_bitmap & (1U << iface_index)) != 0); + CY_ASSERT(iface_index <= CY_UDP_POSIX_IFACE_COUNT_MAX); + CY_ASSERT(is_valid_ip(self->local_ip[iface_index])); + CY_ASSERT(udp_wrapper_is_open(sock)); + CY_ASSERT((self->stats.subject_reader_count == 0) == (self->reader_head == NULL)); + CY_ASSERT((self->stats.subject_reader_count == 0) == (self->reader_tail == NULL)); + CY_ASSERT((reader == NULL) || !reader->tombstone); // Allocate memory that we will read the data into. The ownership of this memory will be transferred // to LibUDPard, which will free it when it is no longer needed. @@ -630,7 +630,7 @@ static void read_socket(cy_udp_posix_t* const self, } // Pass the data buffer into LibUDPard then into Cy for further processing. It takes ownership of the buffer. - assert((reader == NULL) || !reader->port.is_unicast); + CY_ASSERT((reader == NULL) || !reader->port.is_unicast); const bool pushok = udpard_rx_port_push(&self->udpard_rx, (reader != NULL) ? &reader->port : &self->unicast_port, ts, @@ -655,7 +655,7 @@ static cy_err_t spin_once_until(cy_udp_posix_t* const self, const cy_us_t deadli const uint16_t tx_pending_iface_bitmap = udpard_tx_pending_ifaces(&self->udpard_tx); for (uint_fast8_t i = 0; i < CY_UDP_POSIX_IFACE_COUNT_MAX; i++) { if ((tx_pending_iface_bitmap & (1U << i)) != 0) { - assert((self->iface_bitmap & (1U << i)) != 0); + CY_ASSERT((self->iface_bitmap & (1U << i)) != 0); tx_await[tx_count] = &self->sock[i]; tx_count++; } @@ -677,7 +677,7 @@ static cy_err_t spin_once_until(cy_udp_posix_t* const self, const cy_us_t deadli } else { for (uint_fast8_t i = 0; i < CY_UDP_POSIX_IFACE_COUNT_MAX; i++) { if (udp_wrapper_is_open(&rd_iter->sock[i])) { - assert(rx_count < max_rx_count); + CY_ASSERT(rx_count < max_rx_count); rx_await[rx_count] = &rd_iter->sock[i]; rx_readers[rx_count] = rd_iter; rx_iface_indexes[rx_count] = i; @@ -690,7 +690,7 @@ static cy_err_t spin_once_until(cy_udp_posix_t* const self, const cy_us_t deadli // Note that we may add the same socket both for reading and writing, which is fine. for (uint_fast8_t i = 0; i < CY_UDP_POSIX_IFACE_COUNT_MAX; i++) { if (udp_wrapper_is_open(&self->sock[i])) { - assert(rx_count < max_rx_count); + CY_ASSERT(rx_count < max_rx_count); rx_await[rx_count] = &self->sock[i]; rx_readers[rx_count] = NULL; // A unicast socket has no associated topic. rx_iface_indexes[rx_count] = i; @@ -716,7 +716,7 @@ static cy_err_t spin_once_until(cy_udp_posix_t* const self, const cy_us_t deadli if (rx_ready) { self->rx_reverse = !self->rx_reverse; } - assert(res == CY_OK); + CY_ASSERT(res == CY_OK); // While handling the events, we could have generated additional TX items, so we need to process them again. // We do it even in case of failure such that transient errors do not stall the TX queue. // We blindly attempt to transmit on all sockets disregarding their writeability state; if this becomes @@ -793,9 +793,9 @@ static const cy_platform_vtable_t platform_vtable = { static bool v_tx_eject(udpard_tx_t* const tx, udpard_tx_ejection_t* const ej) { cy_udp_posix_t* const self = (cy_udp_posix_t*)tx->user; - assert(self != NULL); - assert((self->iface_bitmap & (1U << ej->iface_index)) != 0); // the caller must ensure this - assert(ej->now <= ej->deadline); + CY_ASSERT(self != NULL); + CY_ASSERT((self->iface_bitmap & (1U << ej->iface_index)) != 0); // the caller must ensure this + CY_ASSERT(ej->now <= ej->deadline); // The libudpard TX API provides us with an opportunity to retain the ownership of the datagram payload // via reference counting. This is useful in kernel space or in embedded systems with low-level NIC drivers, // but the Berkeley socket API does not allow us to take advantage of that -- the data will be copied into the @@ -925,7 +925,7 @@ cy_platform_t* cy_udp_posix_new_manual(const uint64_t uid, } // Finish. - assert(self->unicast_port.is_unicast); + CY_ASSERT(self->unicast_port.is_unicast); self->stats = (cy_udp_posix_stats_t){ 0 }; self->reader_head = NULL; self->reader_tail = NULL; @@ -936,7 +936,7 @@ cy_str_t cy_udp_posix_home(const cy_platform_t* base, const char* const prefix) { char uid[17] = { 0 }; (void)snprintf(uid, sizeof(uid), "%016jx", (uintmax_t)(((cy_udp_posix_t*)base)->udpard_tx.local_uid)); - assert(uid[16] == 0); + CY_ASSERT(uid[16] == 0); static thread_local char g_home[CY_TOPIC_NAME_MAX + 1]; return cy_name_join(cy_str(prefix), cy_str(uid), CY_TOPIC_NAME_MAX, g_home); } @@ -964,9 +964,9 @@ void cy_udp_posix_destroy(cy_platform_t* const base) } rd_iter = next; } - assert(self->stats.subject_reader_count == 0); - assert(self->stats.subject_writer_count == 0); - assert(self->base.cy == NULL); // must be unlinked beforehand + CY_ASSERT(self->stats.subject_reader_count == 0); + CY_ASSERT(self->stats.subject_writer_count == 0); + CY_ASSERT(self->base.cy == NULL); // must be unlinked beforehand udpard_rx_port_free(&self->udpard_rx, &self->unicast_port); udpard_tx_free(&self->udpard_tx); for (uint_fast8_t i = 0; i < CY_UDP_POSIX_IFACE_COUNT_MAX; i++) { diff --git a/cy_udp_posix/udp_wrapper.c b/cy_udp_posix/udp_wrapper.c index 75dfa62..a3756fa 100644 --- a/cy_udp_posix/udp_wrapper.c +++ b/cy_udp_posix/udp_wrapper.c @@ -17,7 +17,6 @@ #include "udp_wrapper.h" -#include #include #include #include @@ -121,7 +120,6 @@ int16_t udp_wrapper_open_unicast(udp_wrapper_t* const self, self->fd = -1; } } - assert((res < 0) || udp_wrapper_is_open(self)); return res; } @@ -178,7 +176,6 @@ int16_t udp_wrapper_open_multicast(udp_wrapper_t* const self, self->fd = -1; } } - assert((res < 0) || udp_wrapper_is_open(self)); return res; }