provider/fsof/FSOFProtocol.php

<?php
/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */
namespace com\fenqile\fsof\provider\fsof;

use com\fenqile\fsof\common\log\FSOFSystemUtil;
use com\fenqile\fsof\common\context\FSOFContext;
use com\fenqile\fsof\common\protocol\fsof\DubboParser;
use com\fenqile\fsof\common\protocol\fsof\DubboRequest;
use com\fenqile\fsof\common\protocol\fsof\DubboResponse;
use com\fenqile\fsof\provider\core\protocol\BufferedProtocol;


class FSOFProtocol extends BufferedProtocol
{
	protected $parser;
    //缓存各consumer请求包数据，直到请求包数据完整接收到
    protected $buffer_header = array();

	//定义swoole work 运行状态
	const FSOF_SWOOLE_STATUS_OK = 0;
	const FSOF_SWOOLE_STATUS_OVERLOAD = 1;
    private $logger;

    public function init()
    {
        $this->logger = \Logger::getLogger(__CLASS__);
        $this->parser = new DubboParser();
    }

    private function checkHeader($client_id, $fsof_data)
    {
    	$request = NULL;
        if (!isset($this->requests[$client_id]))
		{
			//新连接
            $this->logger->debug("new request from {$client_id}");
            if (!empty($this->buffer_header[$client_id]))
			{
                $fsof_data = $this->buffer_header[$client_id].$fsof_data;
            }
            $this->buffer_header[$client_id] = $fsof_data;

            //数据长度还不够
            if (strlen($fsof_data) < DubboParser::PACKAGE_HEDA_LEN)
			{
                return false;
            }
			else
			{
                unset($this->buffer_header[$client_id]);
                $request = new DubboRequest();
                $request->setFullData($fsof_data);
                $request = $this->parser->parseRequestHeader($request);
                //解析失败
                if ($request == false)
				{
                    $this->logger->error("parse request Header fail. fsof_data=" . $fsof_data);
                    return false;
                }
                //保存请求
                $this->logger->debug("create one request for {$client_id}");
                $this->requests[$client_id] = $request;
            }
        }
		else
		{
            $this->logger->debug("append request data for {$client_id}");
            $request = $this->requests[$client_id];
            $request->setFullData($fsof_data);
        }
        return $request;
    }

    public function checkBuffer($client_id, $data)
    {
        //检测头
        $request = $this->checkHeader($client_id, $data);
        //错误的http头
        if ($request === false)
		{
            if (empty($this->buffer_header[$client_id]))
			{
                $this->logger->error("fsof header err.");
                return self::STATUS_ERROR;
            }
			else
			{
                $this->logger->debug("wait head data. fd={$client_id}");
                return self::STATUS_WAIT;
            }
        }

        if ($request->getRequestLen() <= strlen($request->getFullData()))
		{
			if($this->parser->isHearBeatRequest($request))
			{
				//心跳机制
				return self::STATUS_FINISH;
			}
			else
			{
				if ($this->parser->parseRequestBody($request))
				{
                    $this->logger->debug("parse request ok!");
					return self::STATUS_FINISH;
				}
				else
				{
                    $this->logger->error("fsof body err.");
					return self::STATUS_ERROR;
				}
			}
        }
		else
		{
            $this->logger->debug("wait body data. fd={$client_id}");
            return self::STATUS_WAIT;
        }
    }

	private function checkSwooleStatus($request, $host, $port)
	{
		$status = self::FSOF_SWOOLE_STATUS_OK;
		$appConfig = $this->getAppConfig();
		if(false == $appConfig['fsof_setting']['overload_mode'])
		{
			return $status;
		}
		$reqInQueueTime = DubboParser::getReqInQueueTime($request);
		if($reqInQueueTime)
		{
			$appName = $this->getAppName();
			//waiting time inqueue(ms)
			$waitingTime = $appConfig['fsof_setting']['waiting_time'];
			//Number of packet overload in a row
			$overloadNumber = $appConfig['fsof_setting']['overload_number'];
			$lossNumber = $appConfig['fsof_setting']['loss_number'];
            $this->logger->debug($host.":".$port.', waitingTime = '.$waitingTime.', overloadNumber = '.$overloadNumber.', lossNumber = '.$lossNumber);
			$curTimestamp = round(microtime(TRUE)*1000);
			$reqInQueueTime_ms = round($reqInQueueTime / 1000);
			if(($curTimestamp - $reqInQueueTime_ms) >= $waitingTime)
			{
				//swoole_table记录过载的次数新增1
				$this->server->getOverloadMonitor()->overloadIncr();
                $this->logger->warn($host.':'.$port.'|' .$appName.'|服务过载|inQueue:'.$reqInQueueTime_ms.'; curTime:'.$curTimestamp.';waitingTime:'.$waitingTime);
				//判断连续过载次数是否达到开启丢包请求阀值
				if ($this->server->getOverloadMonitor()->getoverloadNum() >= $overloadNumber)
				{
					//重置过载次数,开启过载丢包模式
					$this->server->getOverloadMonitor()->resetOverloadNum_setLossNum($lossNumber);
                    $this->logger->error($host.':'.$port.'|' .$appName.'|服务连续过载,开启丢消息模式');
				}
				$status = self::FSOF_SWOOLE_STATUS_OVERLOAD;
			}
			else
			{
				//清理swoole_table过载记录数量
				$this->server->getOverloadMonitor()->clear();
			}
		}
		return $status;
	}

	private function requestProcessor($client_id, $request)
	{
		//开始执行时间
		$request->startTime = microtime(true);
		//监控请求数量
		$this->server->getAppMonitor()->onRequest($request);

		//设置traceContext, 增加本地的IP地址及APP的端口
		$appConfig = $this->getAppConfig();
		$localIP = FSOFSystemUtil::getLocalIP();
		$appPort = $appConfig['server']['listen'][0];

		$params = $request->__toString();
		if (mb_strlen($params, 'UTF-8') >= 512)
		{
			$params = mb_substr($params, 0, 512, 'UTF-8').' ...';
		}
        $this->logger->debug("in|".$params);

        $businessError = false;
        $frameError = false;
		$result = null;
		//业务处理状态
		$requestFlag = false;
		//返回给客户端执行结果信息
		//$errMsg = 'ok';		//异常信息

		//消息在队列等待时间
		$wait_InQueueTime = 0;
		$inQueueTime = DubboParser::getReqInQueueTime($request);
		if($inQueueTime)
		{
			$wait_InQueueTime = round(microtime(true)*1000000) - $inQueueTime;
		}

		//处理前先检测连接是否仍正常，如己断开则不进行处理
		if(!$this->swoole_server->exist($client_id))
		{
			//执行结束时间
			$request->endTime = microtime(true);
			$cost_time = (int)(($request->endTime - $request->startTime)* 1000000);
			goto END_TCP_CLOSE;
		}

		$status = $this->checkSwooleStatus($request, $localIP, $appPort);
		if(self::FSOF_SWOOLE_STATUS_OK == $status)
		{
			if($this->server->serviceExist($request->getService(),  $request->getGroup(), $request->getVersion()))
			{
				$serviceInstance = $this->server->getServiceInstance($request->getService(), $request->getGroup(), $request->getVersion());
				if (null != $serviceInstance)
				{
					try
					{
						$serviceReflection = new \ReflectionObject($serviceInstance);
						if ($serviceReflection->hasMethod($request->getMethod()))
						{
							$method = $serviceReflection->getmethod($request->getMethod());
							//允许invoke protected方法
							$method->setAccessible(true);
                            $params = $request->getParams();
                            if($params == NULL)
                            {
                                $params = array();
                            }
							$result = $method->invokeArgs($serviceInstance, $params);
							$requestFlag = true;
						}
						else
						{
                            $businessError = true;
							$result = 'function not found:'.$request->getMethod().' in '.$request->getService();
                            $this->logger->error("[{$request->getMethod()}] function not found:".$request->getService());
						}
					}
					catch (\Exception $e)
					{
                        $this->logger->error($e);
                        $frameError = true;
                        $result = $e->getMessage().' in '.$e->getFile().'|'.$e->getLine();
					}

					//如果provider service有状态，则$serviceInstance用完后unset,下次请求重新new, 防止内存泄漏; 对于无状态的service,AppContext会复用$serviceInstance
					if (!$this->server->isStateless())
					{
						unset($serviceInstance);
					}
					unset($method);
					unset($serviceReflection);
				}
				else
				{
                    $frameError = true;
					$result ='get instance failed! | '.$request->getService();
                    $this->logger->error(json_encode($result));
				}
			}
			else
			{
                $frameError = true;
				$result = 'service not found:'.$request->getGroup()."/".$request->getService().":".$request->getVersion();
                $this->logger->error(json_encode($result));
			}
		}
		else
		{
            $frameError = true;
			$result = 'provider过载, 请求消息在队列等待时间超过阀值';
		}

		$request->endTime = microtime(true);//执行结束时间
		$cost_time = (int)(($request->endTime - $request->startTime)* 1000000);
				
		if($this->swoole_server->exist($client_id))
		{
			//发送response
			$response = $this->packResponse($client_id, $request, $result, $businessError,$frameError);
			$msg = $response->__toString();
			if (mb_strlen($msg, 'UTF-8') >= 512)
			{
				$msg = mb_substr($msg, 0, 512, 'UTF-8').' ...('.strlen($msg).')';
			}

            $this->logger->debug(sprintf("out|%s|invokeCostTime:%dus|waitInQueueTime:%dus", $msg, $cost_time, $wait_InQueueTime));
		}
		else
		{
			END_TCP_CLOSE:
			$errMsg = "socket closed by consumer, provider discard response data";
            $this->logger->error("out|{$errMsg}|invokeCostTime:{$cost_time}us| waitInQueueTime:{$wait_InQueueTime}us");
			$requestFlag = false;
		}

		if($requestFlag)
		{
			//监控请求正常处理数量
			$this->server->getAppMonitor()->onResponse($request);
		}
		else
		{
			//监控请求错误处理数量
			$this->server->getAppMonitor()->onError($request);
		}
	}

	public function onOneRequest($client_id, $request)
	{

		if($this->parser->isHearBeatRequest($request))
		{
			//心跳请求,不需要数据回送
		}
		else if($this->parser->isNormalRequest($request) || $this->parser->isOneWayRequest($request))
		{
			//获取配置信息
			$appConfig = $this->getAppConfig();
			
			//Number of packet loss in a row
			$lossNumber = $appConfig['fsof_setting']['loss_number'];
			
			//判断是否开启过载丢包模式
			$restOfLostNum = $this->server->getOverloadMonitor()->getLossNum();
			if($restOfLostNum > 0)
			{
				// 加入对过载丢失数据的统计
				$this->server->getAppMonitor()->onRequest($request);
				
				//回复客户端
				if($restOfLostNum >= $lossNumber)
				{
					if($this->swoole_server->exist($client_id))
					{
						$result = 'provider连续过载, 开启丢消息模式';
						$this->packResponse($client_id, $request, $result, false, true);
					}
				}
				else
				{
					if($this->swoole_server->exist($client_id))
					{
						$result = 'provider开启丢消息模式, 连续丢包中...';
						$this->packResponse($client_id, $request, $result, false, true);
					}
				}
				
				//递减丢包数量
				$this->server->getOverloadMonitor()->lossNumDecr();

				//监控请求错误处理数量
				$this->server->getAppMonitor()->onError($request);
			}
			else
			{
				$this->requestProcessor($client_id,$request);
			}
		}
		else
		{
            $this->logger->error("invalid request = $request");
		}
	}

	public function packResponse($client_id, $request, $data, $businessError,$frameError)
	{
		$response = new DubboResponse();
		$response->setSn($request->getSn());
        if($frameError){
            $response->setStatus(DubboResponse::SERVICE_ERROR);
            $response->setErrorMsg($data);
        }
        if($businessError){
            $response->setErrorMsg($data);
        }
		$response->setResult($data);
		$this->sendResponse($response, $client_id);

		return $response;
	}

    public function sendResponse(DubboResponse $response, $client_id)
    {
        try
		{
            $send_data = $this->parser->packResponse($response);
            $send_len = strlen($send_data);
    
            //默认所有server的response最大5M,每个分包允许重发2次，预计最多10次循环，防止网络出错导致server直接挂死在循环中
            $cnt = (($send_len / DubboParser::RESPONSE_TCP_SEGMENT_LEN) + 1) * 2;
            $tmp_len = $send_len;
            for ($i = 0; $i < $cnt; $i++)
			{
                if ($tmp_len > DubboParser::RESPONSE_TCP_SEGMENT_LEN)
				{
                    //大于1M 分段发送
                    $tmp_data = substr($send_data, 0, DubboParser::RESPONSE_TCP_SEGMENT_LEN);
                    if ($this->swoole_server->send($client_id, $tmp_data))
					{
                        $tmp_len -= DubboParser::RESPONSE_TCP_SEGMENT_LEN;
                        $send_data = substr($send_data, DubboParser::RESPONSE_TCP_SEGMENT_LEN);
                    }
					else
					{
                        if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN')
                        {
                            $last_error_no = $this->swoole_server->errno();
                        }
                        else
                        {
                            $last_error_no = swoole_errno();
                        }
                        if (0 == $last_error_no)
						{
                            //表示该连接己关闭
                            $this->logger->error("当前连接己关闭，发送失败");
                            break;
                        }
						else
						{
                            $this->logger->error('send response split package fail one time!');
                        }
                    }
                    $this->logger->warn('the length of response: '.$send_len.'; send split package '.$i.'/'.$cnt);
                }
				else
				{
                    //小于1M一次性发完
                    if ($this->swoole_server->send($client_id, $send_data))
					{
                        break;
                    }
					else
					{
                        if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN')
                        {
                            $last_error_no = $this->swoole_server->errno();
                        }
                        else
                        {
                            $last_error_no = swoole_errno();
                        }
                        if (0 == $last_error_no)
						{
                            //表示该连接己关闭
                            $this->logger->error("当前连接己关闭，发送失败");
                            break;
                        }
						else
						{
                            $this->logger->error('send response last package fail one time!');
                        }
                    }
                }
            }
        }
		catch (\Exception $e)
		{
            $this->logger->error($e->getMessage(), $e);
        }

        return $send_len;
    }
}