src/btm/btm_thread.c

/*
 * Copyright (C) 2021 Huawei Device Co., Ltd.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "btm_thread.h"

#include <stddef.h>

#include "btstack.h"
#include "platform/include/allocator.h"
#include "platform/include/list.h"
#include "platform/include/mutex.h"
#include "platform/include/queue.h"
#include "platform/include/semaphore.h"
#include "platform/include/thread.h"
#include "log.h"

typedef struct {
    void (*task)(void *context);
    void *context;
} BtmTask;

typedef struct {
    uint8_t id;
    Queue *queue;
    ReactorItem *reactorItem;
} BtmProcessingQueue;

static Thread *g_processingThread = NULL;
static List *g_processingQueueList = NULL;
static Mutex *g_processingQueueLock = NULL;

static BtmTask *AllocTask(void (*task)(void *context), void *context)
{
    BtmTask *block = MEM_MALLOC.alloc(sizeof(BtmTask));
    if (block != NULL) {
        block->task = task;
        block->context = context;
    }
    return block;
}

static void FreeTask(void *task)
{
    MEM_MALLOC.free(task);
}

NO_SANITIZE("cfi") static void RunTask(void *context)
{
    BtmTask *task = QueueTryDequeue((Queue *)context);
    if (task != NULL) {
        task->task(task->context);

        FreeTask(task);
    }
}

static BtmProcessingQueue *AllocProcessingQueue(uint8_t id, uint32_t size)
{
    BtmProcessingQueue *block = MEM_MALLOC.alloc(sizeof(BtmProcessingQueue));
    if (block != NULL) {
        block->id = id;
        block->queue = QueueCreate(size);
        if (block->queue != NULL) {
            Reactor *reactor = ThreadGetReactor(g_processingThread);
            block->reactorItem = ReactorRegister(reactor, QueueGetDequeueFd(block->queue), block->queue, RunTask, NULL);
        }
    }
    return block;
}

typedef struct {
    Queue *queue;
    Semaphore *semaphore;
} RunAllTaskContext;

static void RunAllTaskInQueueTask(void *param)
{
    RunAllTaskContext *context = (RunAllTaskContext *)param;

    BtmTask *task = QueueTryDequeue(context->queue);
    while (task != NULL) {
        task->task(task->context);

        FreeTask(task);

        task = QueueTryDequeue(context->queue);
    }

    if (context->semaphore != NULL) {
        SemaphorePost(context->semaphore);
    }
}

static void RunAllTaskInQueue(Queue *queue)
{
    RunAllTaskContext context = {
        .queue = queue,
        .semaphore = NULL,
    };

    if (ThreadIsSelf(g_processingThread) == 0) {
        RunAllTaskInQueueTask(&context);
    } else {
        context.semaphore = SemaphoreCreate(0);
        ThreadPostTask(g_processingThread, RunAllTaskInQueueTask, &context);
        SemaphoreWait(context.semaphore);
        SemaphoreDelete(context.semaphore);
    }
}

static void FreeProcessingQueue(void *queue)
{
    BtmProcessingQueue *block = (BtmProcessingQueue *)queue;
    if (block->reactorItem != NULL) {
        ReactorUnregister(block->reactorItem);
        block->reactorItem = NULL;
    }
    if (block->queue != NULL) {
        QueueDelete(block->queue, FreeTask);
        block->queue = NULL;
    }
    MEM_MALLOC.free(queue);
}

static BtmProcessingQueue *FindProcessingQueueById(uint8_t queueId)
{
    BtmProcessingQueue *queue = NULL;

    ListNode *node = NULL;
    node = ListGetFirstNode(g_processingQueueList);
    while (node != NULL) {
        queue = ListGetNodeData(node);
        if (queue->id == queueId) {
            break;
        } else {
            queue = NULL;
        }
        node = ListGetNextNode(node);
    }

    return queue;
}

void BtmInitThread()
{
    g_processingThread = ThreadCreate("Stack");
    g_processingQueueList = ListCreate(FreeProcessingQueue);
    g_processingQueueLock = MutexCreate();
}

void BtmCloseThread()
{
    if (g_processingQueueList != NULL) {
        ListDelete(g_processingQueueList);
        g_processingQueueList = NULL;
    }

    if (g_processingThread != NULL) {
        ThreadDelete(g_processingThread);
        g_processingThread = NULL;
    }

    if (g_processingQueueLock != NULL) {
        MutexDelete(g_processingQueueLock);
        g_processingQueueLock = NULL;
    }
}

Thread *BTM_GetProcessingThread()
{
    return g_processingThread;
}

int BTM_CreateProcessingQueue(uint8_t queueId, uint16_t size)
{
    int result = BT_SUCCESS;
    MutexLock(g_processingQueueLock);

    BtmProcessingQueue *queue = FindProcessingQueueById(queueId);
    if (queue != NULL) {
        result = BT_BAD_STATUS;
    } else {
        queue = AllocProcessingQueue(queueId, size);
        ListAddLast(g_processingQueueList, queue);
    }

    MutexUnlock(g_processingQueueLock);
    return result;
}

int BTM_DeleteProcessingQueue(uint8_t queueId)
{
    int result = BT_SUCCESS;

    Queue *taskQueue = NULL;

    MutexLock(g_processingQueueLock);
    BtmProcessingQueue *queue = FindProcessingQueueById(queueId);
    if (queue != NULL) {
        taskQueue = queue->queue;
        queue->queue = NULL;
        ListRemoveNode(g_processingQueueList, queue);
    } else {
        result = BT_BAD_STATUS;
    }
    MutexUnlock(g_processingQueueLock);

    if (taskQueue != NULL) {
        RunAllTaskInQueue(taskQueue);
        QueueDelete(taskQueue, FreeTask);
        taskQueue = NULL;
    }

    return result;
}

int BTM_RunTaskInProcessingQueue(uint8_t queueId, void (*task)(void *context), void *context)
{
    HILOGD("%{public}d ,start process queueId is ", queueId);
    int result = BT_SUCCESS;
    MutexLock(g_processingQueueLock);
    BtmProcessingQueue *queue = FindProcessingQueueById(queueId);
    if (queue != NULL) {
        BtmTask *block = AllocTask(task, context);
        if (task != NULL) {
            QueueEnqueue(queue->queue, block);
        } else {
            result = BT_NO_MEMORY;
        }
    } else {
        result = BT_BAD_STATUS;
    }
    MutexUnlock(g_processingQueueLock);
    HILOGD("%{public}d ,end process queueId is ", queueId);
    return result;
}