Swoole 源码分析之 Channel 通道模块

yxhsea 的个人博客 / 130 / 0 / 创建于 4周前 / 更新于 4周前

原文首发链接：Swoole 源码分析之 Channel 通道模块
大家好，我是码农先森。

引言

通道，用于协程间通讯，支持多生产者协程和多消费者协程。底层自动实现了协程的切换和调度。

通道与 PHP 的 Array 类似，仅占用内存，没有其他额外的资源申请，所有操作均为内存操作，无 IO 消耗。

底层使用 PHP 引用计数实现，无内存拷贝。即使是传递巨大字符串或数组也不会产生额外性能消耗 channel 基于引用计数实现，是零拷贝的。

源码拆解

Channel 通道需要在协程环境中使用，我们先看下面这段代码，使用 new Channel(1) 创建一个 channel 对象，然后在第一个协程中向通道中推送数据，在第二个协程获取到通道内的数据进行消费。

use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use function Swoole\Coroutine\run;

run(function(){
    // 创建 channel 通道对象
    $channel = new Channel(1);
    Coroutine::create(function () use ($channel) {
        for($i = 0; $i < 10; $i++) {
            Coroutine::sleep(1.0);
            // 向通道内推送数据
            $channel->push(['rand' => rand(1000, 9999), 'index' => $i]);
            echo "{$i}\n";
        }
    });
    Coroutine::create(function () use ($channel) {
        while(1) {
            // 从通道中获取数据
            $data = $channel->pop(2.0);
            if ($data) {
                var_dump($data);
            } else {
                assert($channel->errCode === SWOOLE_CHANNEL_TIMEOUT);
                break;
            }
        }
    });
});

在分析源代码之前，我们可以提前看一下源码整体的调用逻辑图，以便我们有个大致的印象。

这段代码主要是在 Swoole 的协程环境中创建 Channel 对象并初始化其容量的逻辑。

// swoole-src/ext-src/swoole-channel.cc:132
static PHP_METHOD(swoole_channel_coro, __construct) {
    zend_long capacity = 1;

    // 解析传入的参数
    ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 0, 1)
    Z_PARAM_OPTIONAL
    Z_PARAM_LONG(capacity)
    ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);

    if (capacity <= 0) {
        capacity = 1;
    }

    // 当前对象对应的 ChannelObject 结构体指针
    ChannelObject *chan_t = php_swoole_channel_coro_fetch_object(Z_OBJ_P(ZEND_THIS));
    // 为该通道对象分配新的 Channel 实例，并设置其容量为传入的值。
    chan_t->chan = new Channel(capacity);
    zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("capacity"), capacity);
}

这段代码主要是在 Swoole 的协程环境中向通道中推送数据并对返回结果进行处理的逻辑。

// swoole-src/ext-src/swoole-channel.cc:149
static PHP_METHOD(swoole_channel_coro, push) {
    // 获取当前对象的 Channel 实例
    Channel *chan = php_swoole_get_channel(ZEND_THIS);
    zval *zdata;
    double timeout = -1;

    // 解析传入的参数
    ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 1, 2)
    Z_PARAM_ZVAL(zdata)
    Z_PARAM_OPTIONAL
    Z_PARAM_DOUBLE(timeout)
    ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);

    Z_TRY_ADDREF_P(zdata);
    zdata = sw_zval_dup(zdata);
    // 向通道中推入数据
    if (chan->push(zdata, timeout)) {
        zend_update_property_long(
            swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), Channel::ERROR_OK);
        RETURN_TRUE;
    } else {
        zend_update_property_long(
            swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), chan->get_error());
        Z_TRY_DELREF_P(zdata);
        efree(zdata);
        RETURN_FALSE;
    }
}

// swoole-src/coroutine/channel.cc:105
bool Channel::push(void *data, double timeout) {
    // 获取当前协程对象 current_co
    Coroutine *current_co = Coroutine::get_current_safe();
    // 如果通道已关闭
    if (closed) {
        // 设置错误并返回空指针
        error_ = ERROR_CLOSED;
        return false;
    }
    // 如果通道已满或生产者队列不为空，则设置超时消息，并根据传入的超时值添加定时器，等待生产者。
    if (is_full() || !producer_queue.empty()) {
        TimeoutMessage msg;
        msg.error = false;
        msg.timer = nullptr;
        if (timeout > 0) {
            msg.chan = this;
            msg.type = PRODUCER;
            msg.co = current_co;
            // 根据传入的超时值添加定时器
            msg.timer = swoole_timer_add(timeout, false, timer_callback, &msg);
        }

        // 挂起生产者协程
        yield(PRODUCER);

        // 如果设置了定时器，则在超时消息中删除定时器
        if (msg.timer) {
            swoole_timer_del(msg.timer);
        }

        // 如果当前协程被取消
        if (current_co->is_canceled()) {
            // 设置错误并返回空指针
            error_ = ERROR_CANCELED;
            return nullptr;
        }

        // 如果发生超时
        if (msg.error) {
            // 设置错误并返回空指针
            error_ = ERROR_TIMEOUT;
            return nullptr;
        }

        // 如果通道关闭且为空的情况
        if (closed && is_empty()) {
            // 设置相应的错误并返回空指针。
            error_ = ERROR_CLOSED;
            return nullptr;
        }
    }

    // 将数据压入数据队列。
    data_queue.push(data);
    swoole_trace_log(SW_TRACE_CHANNEL, "push data to channel, count=%ld", length());

    // 如果消费者队列不为空，则唤醒消费者协程。
    if (!consumer_queue.empty()) {
        Coroutine *co = pop_coroutine(CONSUMER);
        // 恢复消费者协程
        co->resume();
    }
    return true;
}

这段代码主要是在 Swoole 的协程环境中从通道中取出数据并对返回结果进行处理的逻辑。

// swoole-src/ext-src/swoole-channel.cc:175
static PHP_METHOD(swoole_channel_coro, pop) {
    // 获取当前对象的 Channel 实例
    Channel *chan = php_swoole_get_channel(ZEND_THIS);
    // 设置超时变量为-1
    double timeout = -1;

    // 解析一个超时参数
    ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 0, 1)
    Z_PARAM_OPTIONAL
    Z_PARAM_DOUBLE(timeout)
    ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);

    // 从通道中取出数据，并返回一个 zval 指针
    zval *zdata = (zval *) chan->pop(timeout);
    // 如果返回的 zval 指针不为空
    if (zdata) {
        // 将其返回给 PHP 脚本，并释放内存
        RETVAL_ZVAL(zdata, 0, 0);
        efree(zdata);
        zend_update_property_long(
            swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), Channel::ERROR_OK);
    } else {
        zend_update_property_long(
            swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), chan->get_error());
        RETURN_FALSE;
    }
}

// swoole-src/coroutine/channel.cc:55
void *Channel::pop(double timeout) {
    // 获取当前协程对象 current_co
    Coroutine *current_co = Coroutine::get_current_safe();
    // 如果通道已关闭且为空
    if (closed && is_empty()) {
        // 设置错误并返回空指针
        error_ = ERROR_CLOSED;
        return nullptr;
    }
    // 如果通道为空或者消费者队列不为空
    if (is_empty() || !consumer_queue.empty()) {
        TimeoutMessage msg;
        msg.error = false;
        msg.timer = nullptr;
        if (timeout > 0) {
            msg.chan = this;
            msg.type = CONSUMER;
            msg.co = current_co;
            // 根据传入的超时值添加定时器
            msg.timer = swoole_timer_add(timeout, false, timer_callback, &msg);
        }

        // 挂起消费者协程
        yield(CONSUMER);

        // 如果设置了定时器，则在超时消息中删除定时器
        if (msg.timer) {
            swoole_timer_del(msg.timer);
        }

        // 如果当前协程被取消
        if (current_co->is_canceled()) {
            // 设置错误并返回空指针
            error_ = ERROR_CANCELED;
            return nullptr;
        }

        // 如果发生超时
        if (msg.error) {
            // 设置错误并返回空指针
            error_ = ERROR_TIMEOUT;
            return nullptr;
        }

        // 如果通道关闭且为空的情况
        if (closed && is_empty()) {
            // 设置相应的错误并返回空指针。
            error_ = ERROR_CLOSED;
            return nullptr;
        }
    }

    // 从数据队列中弹出数据，并返回该数据。
    void *data = data_queue.front();
    data_queue.pop();

    // 如果生产者队列不为空，则唤醒生产者协程
    if (!producer_queue.empty()) {
        Coroutine *co = pop_coroutine(PRODUCER);
        // 恢复到生产者协程
        co->resume();
    }
    return data;
}

这段代码一是针对超时回调处理的处理逻辑，并恢复相关的协程操作。二是实现了协程的挂起操作，并根据不同的类型将当前协程放入不同的队列中，以便后续根据需要恢复执行。

// swoole-src/coroutine/channel.cc:22
void Channel::timer_callback(Timer *timer, TimerNode *tnode) {
    TimeoutMessage *msg = (TimeoutMessage *) tnode->data;
    msg->error = true;
    msg->timer = nullptr;
    if (msg->type == CONSUMER) {
        // 从消费者队列中移除该协程
        msg->chan->consumer_remove(msg->co);
    } else {
        // 从生产者队列中移除该协程
        msg->chan->producer_remove(msg->co);
    }
    // 恢复协程
    msg->co->resume();
}

// swoole-src/coroutine/channel.cc:34
void Channel::yield(enum Opcode type) {
    // 获取当前协程
    Coroutine *co = Coroutine::get_current_safe();
    if (type == PRODUCER) {
        // 将当前协程放入到生产者队列
        producer_queue.push_back(co);
        swoole_trace_log(SW_TRACE_CHANNEL, "producer cid=%ld", co->get_cid());
    } else {
        // 将当前协程放入到消费者队列
        consumer_queue.push_back(co);
        swoole_trace_log(SW_TRACE_CHANNEL, "consumer cid=%ld", co->get_cid());
    }

    // 挂起被取消，则调用该函数
    Coroutine::CancelFunc cancel_fn = [this, type](Coroutine *co) {
        if (type == CONSUMER) {
            consumer_remove(co);
        } else {
            producer_remove(co);
        }
        co->resume();
        return true;
    };

    // 挂起当前协程
    co->yield(&cancel_fn);
}

总结

Channel 通道需要在协程的环境中进行使用，通道是纯内存操作，没有 IO 消耗，非常高效。
底层使用 Channel::yield 函数实现了协程的自动切换和调度，如果通道处理超时则会自动调用 Channel::timer_callback 函数。
Channel 通道是跨协程直接通信的一大利器，在实际的场景中使用起来十分的便利、高效。