finish调用导致内存溢出

[diida]

描述:

在异步server 的 onTask的处理中，如果调用finish来处理task结束，每次会增加48个字节（memory_get_usage() 不带true参数），长时间运行便会内存增长到一个非常大的值。

同样的代码，使用return来处理task结束，不会内存溢出

代码:

<?php
/**
 * Created by JetBrains PhpStorm.
 * User: kevin
 * Date: 15-8-4
 * Time: 下午10:14
 * To change this template use File | Settings | File Templates.
 */

namespace Base\cassandraProxyServer;
include_once('config.php');
include_once('cassandra.php');

class server
{
    /**
     * 在send之前调用，将数据长度追加到数据中
     * @param $data
     * @return string
     */
    static function mypack($data)
    {
        $data = \igbinary_serialize($data);
        return pack('N', strlen($data)) . $data;
    }

    /**
     * 在receive后，解压原始数据
     * @param $data
     * @return bool
     */
    static function myunpack($data)
    {
        if (empty($data)) {
            return false;
        }

        $length = unpack("N", $data)[1];
        $msg = substr($data, -$length);
        if ($msg == '0') {
            return true;
        } else {
            $result = \igbinary_unserialize($msg);
            return $result;
        }
    }

    /**
     * 接收到数据后，解压，交给task异步处理
     * @param $serv
     * @param $fd
     * @param $from_id
     * @param $data
     */
    static function onReceive($serv, $fd, $from_id, $data)
    {
        $data = self::myunpack($data);
        $data['fd'] = $fd;
        if (isset($data['server'])) {
            $serv->send($fd, self::mypack([100, $data['sn'], $serv->stats()]));
        } else {
            $serv->task($data);
        }
    }

    /**
     * 处理数据和连接池
     * @param $serv
     * @param $task_id
     * @param $from_id
     * @param $data
     * @return array
     */
    static function onTask($serv, $task_id, $from_id, $data)
    {
        static $cass = null;
        //echo memory_get_usage(), "\n";
        if ($cass == null) {
            $cass = new cassandra();
        }

        if (!empty($data['async'])) {
            //先返回0
            $serv->send($data['fd'], self::mypack('0'));
            list($link, $stmt, $options, $result) = $cass->execute($data, $serv);
            $future = $cass->execASync($link, $stmt, $options);
            //阻塞调用
            //$future->get(2);
            return $result;
        } else {
            return $cass->execute($data, $serv);
        }
    }

    /**
     * 完成task
     * @param $serv
     * @param $taskId
     * @param $result
     */
    static function onFinish($serv, $taskId, $result)
    {
        $fd = array_shift($result);
        if ($result[0] == 120) {
            return;
        }

        $response = self::mypack($result);
        $serv->send($fd, $response);
        //  echo "AsyncTask Finish:Connect.PID=" . posix_getpid() . PHP_EOL;
    }

    static function onClose($server, $fd, $from_id)
    {

    }

    static function start()
    {
        $serv = new \swoole_server(config::getListenAddr(), config::$LISTEN_PORT);
        $serv->set(array(
            'worker_num' => config::$WORKER_NUM,
            'task_worker_num' => config::$TASK_NUM, //Cassandra连接的数量
            'dispatch_mode' => 3, //抢占模式处理请求
            'open_length_check' => true, //在包头增加长度，检查包是否传完，有利于提高效率
            'package_length_type' => 'N', //长度字段的类型，固定包头中用一个4字节表示包体长度。
            'package_max_length' => 2000000, //协议最大长度
            'package_length_offset' => 0, //第N个字节是包长度的值
            'package_body_offset' => 4, //第几个字节开始计算长度
            'backlog' => 128, //队列数量
            'max_conn' => config::$MAX_CONNECT //连接数
        ));

        $serv->on('Receive', ['\\Base\\cassandraProxyServer\\server', 'onReceive']);
        $serv->on('Task', ['\\Base\\cassandraProxyServer\\server', 'onTask']);
        $serv->on('Finish', ['\\Base\\cassandraProxyServer\\server', 'onFinish']);
        $serv->on('Close', ['\\Base\\cassandraProxyServer\\server', 'onClose']);
        $serv->start();
    }
}

server::start();

版本:

1.7.18

环境:

MacOS/CentOS均会出现这个问题

[matyhtf]

已修复，感谢反馈。