Merge remote-tracking branch 'origin/master' into upstream_master

Author: garryyan

mars/comm/network/getdnssvraddrs.cc

@@ -58,9 +58,6 @@ void getdnssvraddrs(std::vector<socket_address>& _dnssvraddrs) {
     res_ndestroy(&stat);
 }
 
-#elif defined WP8
-void getdnssvraddrs(std::vector<socket_address>& _dnssvraddrs) {
-}
 #elif defined _WIN32
 #include <stdio.h>
 #include <windows.h>
@@ -86,5 +83,6 @@ void getdnssvraddrs(std::vector<socket_address>& _dnssvraddrs) {
     return;
 }
 #else
-#error "no imp"
+void getdnssvraddrs(std::vector<socket_address>& _dnssvraddrs) {
+}
 #endif

mars/comm/network/getifaddrs.h

@@ -22,6 +22,7 @@
 
 #include <string>
 #include <vector>
+#include <string.h>
 
 struct ifaddrinfo_ipv4_t
 {

mars/comm/socket/socket_address.h

@@ -23,7 +23,7 @@
 
 #include "unix_socket.h"
 
-#include <string>
+#include <string.h>
 
 class socket_address {
 

mars/comm/socket/tcp_fsm_handler.h

@@ -26,6 +26,7 @@
 template<class InputIterator>
 bool TcpFSMHandler(InputIterator _first, InputIterator _last, SocketSelectBreaker& _breaker, bool _select_anyway)
 {
+	xverbose_function();
     xgroup2_define(group);
 
     SocketSelect sel(_breaker, true);

mars/comm/socket/tcpclient.cc

@@ -180,6 +180,11 @@ void TcpClient::__Run() {
 
     int ret = ::connect(socket_, (sockaddr*)&_addr, sizeof(_addr));
 
+    std::string local_ip = socket_address::getsockname(socket_).ip();
+    unsigned int local_port = socket_address::getsockname(socket_).port();
+
+    xinfo2(TSF"sock:%_, local_ip:%_, local_port:%_, svr_ip:%_, svr_port:%_", socket_, local_ip, local_port, ip_, port_);
+
     if (0 > ret && !IS_NOBLOCK_CONNECT_ERRNO(socket_errno)) {
         xerror2("connect errno=%d", socket_errno);
         status_ = kTcpConnectingErr;

mars/comm/socket/tcpserver_fsm.cc

@@ -26,7 +26,7 @@
 #include "comm/socket/tcpserver_fsm.h"
 
 TcpServerFSM::TcpServerFSM(SOCKET _socket)
-    : status_(kAccept), sock_(_socket) {
+    : status_(kAccept), sock_(_socket), is_write_fd_set_(false) {
     xassert2(INVALID_SOCKET != sock_);
     socklen_t addr_len = sizeof(addr_);
     xerror2_if(0 > getpeername(sock_, (sockaddr*)&addr_, &addr_len), TSF"getpeername:%_, %_", socket_errno, socket_strerror(socket_errno));
@@ -36,7 +36,7 @@ TcpServerFSM::TcpServerFSM(SOCKET _socket)
 }
 
 TcpServerFSM::TcpServerFSM(SOCKET _socket, const sockaddr_in& _addr)
-    : status_(kAccept), sock_(_socket), addr_(_addr) {
+    : status_(kAccept), sock_(_socket), addr_(_addr) , is_write_fd_set_(false){
     memset(ip_, 0, sizeof(ip_));
     inet_ntop(addr_.sin_family, &(addr_.sin_addr), ip_, sizeof(ip_));
 }
@@ -120,7 +120,12 @@ TcpServerFSM::TSocketStatus TcpServerFSM::PreReadWriteSelect(SocketSelect& _sel,
     _sel.Read_FD_SET(sock_);
     _sel.Exception_FD_SET(sock_);
 
-    if (0 < send_buf_.Length())_sel.Write_FD_SET(sock_);
+    if (0 < send_buf_.Length()) {
+    	WriteFDSet(true);
+    	_sel.Write_FD_SET(sock_);
+    } else {
+    	WriteFDSet(false);
+    }
 
     return status_;
 }

mars/comm/socket/tcpserver_fsm.h

@@ -23,6 +23,9 @@
 
 #include "comm/autobuffer.h"
 #include "comm/socket/unix_socket.h"
+#include "comm/xlogger/xlogger.h"
+#include "comm/thread/mutex.h"
+#include "comm/thread/lock.h"
 
 class XLogger;
 class SocketSelect;
@@ -51,6 +54,17 @@ class TcpServerFSM {
     size_t SendBufLen() {return send_buf_.Length();}
     void Close(bool _notify = true);
 
+    bool WriteFDSet()
+    {
+    	ScopedLock lock (write_fd_set_mutex_);
+    	return is_write_fd_set_;
+    }
+    void WriteFDSet(bool _is_set) {
+    	xverbose_function(TSF"_is_set:%_, is_write_fd_set_:%_", _is_set, is_write_fd_set_);
+    	ScopedLock lock (write_fd_set_mutex_);
+    	is_write_fd_set_  = _is_set;
+    }
+
     virtual TSocketStatus PreSelect(SocketSelect& _sel, XLogger& _log);
     virtual TSocketStatus AfterSelect(SocketSelect& _sel, XLogger& _log);
     virtual int Timeout() const;
@@ -80,6 +94,9 @@ class TcpServerFSM {
 
     AutoBuffer send_buf_;
     AutoBuffer recv_buf_;
+
+    bool is_write_fd_set_;
+    Mutex write_fd_set_mutex_;
 };
 
 #endif /* TcpServerFSM_H_ */

mars/comm/unix/thread/thread.h

@@ -373,8 +373,10 @@ class Thread {
         if (0 < strnlen((const char*)runableref->thread_name, sizeof(runableref->thread_name))) {
 #ifdef __APPLE__
             pthread_setname_np((const char*)runableref->thread_name);
-#else
+#elif defined(ANDROID)
             pthread_setname_np(runableref->tid, (const char*)runableref->thread_name);
+#else
+            
 #endif
         }
 

mars/libraries/mars_android_sdk/jni/log_crypt.cc.rewriteme

@@ -24,6 +24,7 @@
 #include <string.h>
 #include <sys/time.h>
 #include <time.h>
+#include <stdio.h>
 
 static const char kMagicSyncStart = '\x03';
 static const char kMagicAsyncStart ='\x05';

mars/libraries/mars_android_sdk/jni/longlink_packer.h

@@ -22,6 +22,7 @@
 #define STN_SRC_LONGLINK_PACKER_H_
 
 #include <stdlib.h>
+#include <stdint.h>
 
 #define LONGLINK_UNPACK_CONTINUE (-2)
 #define LONGLINK_UNPACK_FALSE (-1)

mars/libraries/mars_utils.py

@@ -24,9 +24,9 @@
 WIN_COPY_EXT_FILES = {"stn/proto/longlink_packer.h": "longlink_packer.h",
         "stn/proto/stnproto_logic.h": "stnproto_logic.h",
         "stn/proto/longlink_packer.cc__": "longlink_packer.cc.rewriteme",
-        "comm/windows/projdef.h": "mars-open/comm/projdef.h",
-        "comm/windows/sys/cdefs.h": "mars-open/comm/sys/cdefs.h",
-        "comm/windows/sys/time.h": "mars-open/comm/sys/time.h"
+        "comm/windows/projdef.h": "mars/comm/projdef.h",
+        "comm/windows/sys/cdefs.h": "mars/comm/sys/cdefs.h",
+        "comm/windows/sys/time.h": "mars/comm/sys/time.h"
         }
 
 html_css = '''

mars/log/crypt/log_crypt.cc

@@ -24,6 +24,7 @@
 #include <string.h>
 #include <sys/time.h>
 #include <time.h>
+#include <stdio.h>
 
 static const char kMagicSyncStart = '\x03';
 static const char kMagicAsyncStart ='\x05';

mars/stn/proto/longlink_packer.h

@@ -22,6 +22,7 @@
 #define STN_SRC_LONGLINK_PACKER_H_
 
 #include <stdlib.h>
+#include <stdint.h>
 
 #define LONGLINK_UNPACK_CONTINUE (-2)
 #define LONGLINK_UNPACK_FALSE (-1)

mars/stn/src/longlink.cc

@@ -724,8 +724,7 @@ void LongLink::__RunReadWrite(SOCKET _sock, ErrCmdType& _errtype, int& _errcode,
             if (nwrite_size <= (maxnwrite + it->writelen)) {
                 xinfo2(TSF"taskid:%_, cmdid:%_, taskid:%_ ; ", it->taskid, it->cmdid, it->taskid) >> close_log;
                 break;
-            }
-            else {
+            } else {
                 maxnwrite += it->writelen;
                 xinfo2(TSF"taskid:%_, cmdid:%_, task_info:%_ ; ", it->taskid, it->cmdid, it->task_info) >> close_log;
             }
@@ -753,8 +752,7 @@ void LongLink::__RunReadWrite(SOCKET _sock, ErrCmdType& _errtype, int& _errcode,
 				xinfo2(TSF"taskid:%_, cmdid:%_, task_info:%_; ", taskid, cmdid, sent_taskids[taskid]) >> close_log;
 				if (LONGLINK_UNPACK_CONTINUE == unpackret || LONGLINK_UNPACK_FALSE == unpackret) {
 					break;
-				}
-				else {
+				} else {
 					sent_taskids.erase(taskid);
 					bufrecv.Move(-(int)(packlen));
 				}
@@ -766,7 +764,7 @@ void LongLink::__RunReadWrite(SOCKET _sock, ErrCmdType& _errtype, int& _errcode,
     struct tcp_info _info;
     if (getsocktcpinfo(_sock, &_info) == 0) {
     	char tcp_info_str[1024] = {0};
-    		xinfo2(TSF"task socket close getsocktcpinfo:%_", tcpinfo2str(&_info, tcp_info_str, sizeof(tcp_info_str))) >> close_log;
+        xinfo2(TSF"task socket close getsocktcpinfo:%_", tcpinfo2str(&_info, tcp_info_str, sizeof(tcp_info_str))) >> close_log;
     }
 #endif
 }

mars/stn/src/net_core.cc

@@ -25,9 +25,12 @@
 #include "boost/bind.hpp"
 #include "boost/ref.hpp"
 
+#include "openssl/export_include/openssl_multi_thread_support.h"
+
 #include "mars/app/app.h"
 #include "mars/baseevent/active_logic.h"
 #include "mars/comm/messagequeue/message_queue.h"
+#include "mars/comm/network/netinfo_util.h"
 #include "mars/comm/socket/local_ipstack.h"
 #include "mars/comm/xlogger/xlogger.h"
 #include "mars/comm/singleton.h"
@@ -51,8 +54,7 @@
 
 #include "signalling_keeper.h"
 #include "zombie_task_manager.h"
-#include "openssl/export_include/openssl_multi_thread_support.h"
-#include "comm/network/netinfo_util.h"
+
 using namespace mars::stn;
 using namespace mars::app;
 
@@ -100,13 +102,6 @@ inline  static bool __ValidAndInitDefault(Task& _task, XLogger& _group) {
 
 static const int kShortlinkErrTime = 3;
 
-enum {
-    kCallFromLong,
-    kCallFromShort,
-    kCallFromZombie,
-};
-
-
 
 NetCore::NetCore()
     : messagequeue_creater_(true, XLOGGER_TAG)
@@ -220,8 +215,7 @@ NetCore::~NetCore() {
 #ifdef USE_LONG_LINK
     GetSignalOnNetworkDataChange().disconnect(boost::bind(&SignallingKeeper::OnNetWorkDataChanged, signalling_keeper_, _1, _2, _3));
 
-    longlink_task_manager_->LongLinkChannel().SignalConnection.disconnect(boost::bind(&TimingSync::OnLongLinkStatuChanged, timing_sync_, _1));
-    longlink_task_manager_->LongLinkChannel().SignalConnection.disconnect(boost::bind(&NetCore::__OnLongLinkConnStatusChange, this, _1));
+    longlink_task_manager_->LongLinkChannel().SignalConnection.disconnect_all_slots();
     longlink_task_manager_->LongLinkChannel().broadcast_linkstatus_signal_.disconnect_all_slots();
 
     push_preprocess_signal_.disconnect_all_slots();
@@ -451,6 +445,14 @@ void NetCore::OnNetworkChange() {
    ASYNC_BLOCK_END
 }
 
+void NetCore::KeepSignal() {
+	signalling_keeper_->Keep();
+}
+
+void NetCore::StopSignal() {
+	signalling_keeper_->Stop();
+}
+
 #ifdef USE_LONG_LINK
 #ifdef __APPLE__
 void NetCore::__ResetLongLink() {
@@ -497,6 +499,12 @@ bool NetCore::LongLinkIsConnected() {
 }
 
 int NetCore::__CallBack(int _from, ErrCmdType _err_type, int _err_code, int _fail_handle, const Task& _task, unsigned int _taskcosttime) {
+
+	if (task_callback_hook_ && 0 == task_callback_hook_(_from, _err_type, _err_code, _fail_handle, _task)) {
+		xwarn2(TSF"task_callback_hook let task return. taskid:%_, cgi%_.", _task.taskid, _task.cgi);
+		return 0;
+	}
+
     if (kEctOK == _err_type || kTaskFailHandleTaskEnd == _fail_handle)
     	return OnTaskEnd(_task.taskid, _task.user_context, _err_type, _err_code);
 

mars/stn/src/net_core.h

@@ -57,12 +57,19 @@ class NetCheckLogic;
 class DynamicTimeout;
 class AntiAvalanche;
 
+enum {
+    kCallFromLong,
+    kCallFromShort,
+    kCallFromZombie,
+};
+
 class NetCore {
   public:
     SINGLETON_INTRUSIVE(NetCore, new NetCore, NetCore::__Release);
 
   public:
     boost::function<void (Task& _task)> task_process_hook_;
+    boost::function<int (int _from, ErrCmdType _err_type, int _err_code, int _fail_handle, const Task& _task)> task_callback_hook_;
     boost::signals2::signal<void (uint32_t _cmdid, const AutoBuffer& _buffer)> push_preprocess_signal_;
 
   public:
@@ -76,11 +83,11 @@ class NetCore {
     bool    LongLinkIsConnected();
     void    OnNetworkChange();
 
+    void	KeepSignal();
+    void	StopSignal();
+
 #ifdef USE_LONG_LINK
-    SignallingKeeper&    GetSignallingKeeper() {return *signalling_keeper_;}
     LongLinkTaskManager& GetLongLinkTaskManager() {return *longlink_task_manager_;}
-    NetSourceTimerCheck& GetNetSourceTimerCheck() {return *netsource_timercheck_;}
-    ShortLinkTaskManager& GetShortLinkTaskManager() {return *shortlink_task_manager_;}
 #endif
 
   private:

mars/stn/stn.h

@@ -44,6 +44,7 @@ struct Task {
 
     static const int kChannelNormalStrategy = 0;
     static const int kChannelFastStrategy = 1;
+    static const int kChannelDisasterRecoveryStategy = 2;
 
     static const int kTaskPriorityHighest = 0;
     static const int kTaskPriority0 = 0;

mars/stn/stn_logic.cc

@@ -231,13 +231,13 @@ void SetSignallingStrategy(long _period, long _keepTime) {
 
 void KeepSignalling() {
 #ifdef USE_LONG_LINK
-    STN_WEAK_CALL(GetSignallingKeeper().Keep());
+    STN_WEAK_CALL(KeepSignal());
 #endif
 }
 
 void StopSignalling() {
 #ifdef USE_LONG_LINK
-    STN_WEAK_CALL(GetSignallingKeeper().Stop());
+    STN_WEAK_CALL(StopSignal());
 #endif
 }