* This file is part of the oGRAC project.
* Copyright (c) 2024 Huawei Technologies Co.,Ltd.
*
* oGRAC is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*
* http://license.coscl.org.cn/MulanPSL2
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
* -------------------------------------------------------------------------
*
* cm_chan.c
*
*
* IDENTIFICATION
* src/common/cm_chan.c
*
* -------------------------------------------------------------------------
*/
#include "cm_common_module.h"
#include "cm_chan.h"
#include "cm_error.h"
chan_t *cm_chan_new(uint32 capacity, uint32 size)
{
errno_t rc_memzero;
uint32 real_size;
if (capacity == 0 || size == 0) {
return NULL;
}
chan_t *chan = (chan_t *)malloc(sizeof(*chan));
if (chan == NULL) {
return NULL;
}
rc_memzero = memset_sp(chan, sizeof(*chan), 0, sizeof(*chan));
if (rc_memzero != EOK) {
CM_FREE_PTR(chan);
return NULL;
}
chan->capacity = capacity;
chan->count = 0;
chan->size = size;
real_size = size * capacity;
if (real_size / capacity != size) {
CM_FREE_PTR(chan);
return NULL;
}
chan->buf = (uint8 *)malloc(real_size);
if (chan->buf == NULL) {
CM_FREE_PTR(chan);
return NULL;
}
rc_memzero = memset_sp(chan->buf, (size_t)real_size, 0, (size_t)real_size);
if (rc_memzero != EOK) {
CM_FREE_PTR(chan->buf);
CM_FREE_PTR(chan);
return NULL;
}
chan->buf_end = chan->buf + size * (capacity);
chan->begin = chan->buf;
chan->end = chan->buf;
chan->lock = 0;
(void)cm_event_init(&chan->event_send);
(void)cm_event_init(&chan->event_recv);
chan->waittime_ms = 100;
chan->is_closed = OG_FALSE;
chan->ref_count = 0;
return chan;
}
status_t cm_chan_send_timeout(chan_t *chan, const void *elem, uint32 timeout_ms)
{
errno_t errcode;
if (chan == NULL || elem == NULL) {
return OG_ERROR;
}
cm_spin_lock(&chan->lock, NULL);
{
if (chan->buf == NULL || chan->is_closed) {
cm_spin_unlock(&chan->lock);
return OG_ERROR;
}
while (chan->count == chan->capacity) {
cm_spin_unlock(&chan->lock);
if (OG_TIMEDOUT == cm_event_timedwait(&chan->event_recv, timeout_ms)) {
return OG_TIMEDOUT;
}
cm_spin_lock(&chan->lock, NULL);
if (chan->count < chan->capacity) {
break;
}
}
if (chan->end >= chan->buf_end) {
chan->end = chan->buf;
}
if (chan->size != 0) {
errcode = memcpy_sp(chan->end, (size_t)(chan->buf_end - chan->end), elem, (size_t)chan->size);
if (errcode != EOK) {
cm_spin_unlock(&chan->lock);
OG_THROW_ERROR(ERR_SYSTEM_CALL, errcode);
return OG_ERROR;
}
}
chan->end += chan->size;
chan->count++;
}
cm_spin_unlock(&chan->lock);
cm_event_notify(&chan->event_send);
return OG_SUCCESS;
}
status_t cm_chan_send(chan_t *chan, const void *elem)
{
return cm_chan_send_timeout(chan, elem, 0xFFFFFFFF);
}
status_t cm_chan_recv_timeout(chan_t *chan, void *elem, uint32 timeout_ms)
{
errno_t errcode;
if (chan == NULL || elem == NULL) {
return OG_ERROR;
}
cm_spin_lock(&chan->lock, NULL);
{
if (chan->buf == NULL) {
cm_spin_unlock(&chan->lock);
return OG_ERROR;
}
while (chan->count == 0) {
if (chan->is_closed) {
cm_spin_unlock(&chan->lock);
return OG_ERROR;
}
cm_spin_unlock(&chan->lock);
if (OG_TIMEDOUT == cm_event_timedwait(&chan->event_send, timeout_ms)) {
return OG_TIMEDOUT;
}
cm_spin_lock(&chan->lock, NULL);
if (chan->count > 0) {
break;
}
}
if (chan->begin >= chan->buf_end) {
chan->begin = chan->buf;
}
if (chan->size != 0) {
errcode = memcpy_sp(elem, (size_t)chan->size, chan->begin, (size_t)chan->size);
if (errcode != EOK) {
cm_spin_unlock(&chan->lock);
OG_THROW_ERROR(ERR_SYSTEM_CALL, errcode);
return OG_ERROR;
}
}
chan->begin += chan->size;
chan->count--;
}
cm_spin_unlock(&chan->lock);
cm_event_notify(&chan->event_recv);
return OG_SUCCESS;
}
status_t cm_chan_recv(chan_t *chan, void *elem)
{
return cm_chan_recv_timeout(chan, elem, 0xFFFFFFFF);
}
bool32 cm_chan_empty(chan_t *chan)
{
cm_spin_lock(&chan->lock, NULL);
if (chan->count == 0) {
cm_spin_unlock(&chan->lock);
return OG_TRUE;
}
cm_spin_unlock(&chan->lock);
return OG_FALSE;
}
void cm_chan_close(chan_t *chan)
{
if (chan == NULL) {
return;
}
cm_spin_lock(&chan->lock, NULL);
if (chan->is_closed) {
cm_spin_unlock(&chan->lock);
return;
}
chan->is_closed = OG_TRUE;
uint32 i = 0;
for (i = 0; i < chan->ref_count; i++) {
cm_event_notify(&chan->event_recv);
cm_event_notify(&chan->event_send);
}
cm_spin_unlock(&chan->lock);
}
void cm_chan_free(chan_t *chan)
{
if (chan == NULL) {
return;
}
cm_event_destory(&chan->event_recv);
cm_event_destory(&chan->event_send);
CM_FREE_PTR(chan->buf);
chan->begin = NULL;
chan->end = NULL;
chan->buf_end = NULL;
chan->capacity = 0;
chan->count = 0;
chan->size = 0;
chan->is_closed = OG_TRUE;
chan->ref_count = 0;
CM_FREE_PTR(chan);
}