From 4645013bb26db5373d213ff8d5815d6da4da53a3 Mon Sep 17 00:00:00 2001 From: Ding Li Date: Thu, 8 Jun 2017 18:35:32 +0800 Subject: [PATCH] Fix GlobalHandle bug. Add rebuild method to ZmqMsg. --- Include/Zmq/Context.mqh | 10 ++-- Include/Zmq/GlobalHandle.mqh | 17 +++--- Include/Zmq/ZmqMsg.mqh | 27 +++++++-- README.md | 2 + .../Chapter1/TaskEvent.mq4 | Bin 0 -> 4596 bytes .../ZeroMQGuideExamples/Chapter1/TaskSink.mq4 | 48 ++++++++++++++++ .../Chapter1/TaskWorker.mq4 | 52 ++++++++++++++++++ 7 files changed, 137 insertions(+), 19 deletions(-) create mode 100644 Scripts/ZeroMQGuideExamples/Chapter1/TaskEvent.mq4 create mode 100644 Scripts/ZeroMQGuideExamples/Chapter1/TaskSink.mq4 create mode 100644 Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 diff --git a/Include/Zmq/Context.mqh b/Include/Zmq/Context.mqh index 51d1262..920667d 100644 --- a/Include/Zmq/Context.mqh +++ b/Include/Zmq/Context.mqh @@ -48,16 +48,18 @@ int zmq_ctx_get(intptr_t context,int option); //| and in a manner not easily recognized by humans, for example: | //| "__3kewducdxhkd__" | //+------------------------------------------------------------------+ -class Context: public GlobalHandle +class Context: public GlobalHandle { protected: int get(int option) {return zmq_ctx_get(m_ref,option);} bool set(int option,int optval) {return 0==zmq_ctx_set(m_ref,option,optval);} - intptr_t create() override {return zmq_ctx_new();} - void destroy(intptr_t handle) override {if(0!=zmq_ctx_term(handle)) {Debug("failed to terminate context");}} public: - Context(string shared=NULL):GlobalHandle(shared) {} + + static intptr_t create() {return zmq_ctx_new();} + static void destroy(intptr_t handle) {if(0!=zmq_ctx_term(handle)) {Debug("failed to terminate context");}} + + Context(string shared=NULL):GlobalHandle(shared) {} bool shutdown() {return 0==zmq_ctx_shutdown(m_ref);} diff --git a/Include/Zmq/GlobalHandle.mqh b/Include/Zmq/GlobalHandle.mqh index 73c3065..3de03b5 100644 --- a/Include/Zmq/GlobalHandle.mqh +++ b/Include/Zmq/GlobalHandle.mqh @@ -55,14 +55,15 @@ public: bool isValid() const {return m_name!=NULL;} string getName() const {return m_name;} - void enter() { while(!GlobalVariable::makeTemp(m_name))Sleep(100); } + void enter() { while(!GlobalVariable::makeTemp(m_name) && !IsStopped())Sleep(100); } bool tryEnter() { return GlobalVariable::makeTemp(m_name); } void leave() { GlobalVariable::remove(m_name);} }; //+------------------------------------------------------------------+ //| A reference counted global pointer (or handle) | +//| HandleManager should implement 2 static methods: create & destroy| //+------------------------------------------------------------------+ -template +template class GlobalHandle { private: @@ -76,7 +77,7 @@ public: { m_refName=m_cs.getName()+"_Ref"; m_counterName=m_cs.getName()+"_Count"; - if(!m_cs.isValid()) m_ref=create(); + if(!m_cs.isValid()) m_ref=HandleManager::create(); else { m_cs.enter(); @@ -87,7 +88,7 @@ public: } if(long(GlobalVariable::get(m_counterName))==0) { - m_ref=create(); + m_ref=HandleManager::create(); if(!GlobalVariable::exists(m_refName)) { GlobalVariable::makeTemp(m_refName); @@ -104,12 +105,12 @@ public: } ~GlobalHandle() { - if(!m_cs.isValid()) {destroy(m_ref); return;} + if(!m_cs.isValid()) {HandleManager::destroy(m_ref); return;} m_cs.enter(); GlobalVariable::set(m_counterName,GlobalVariable::get(m_counterName)-1); if(long(GlobalVariable::get(m_counterName))==0) { - destroy(m_ref); + HandleManager::destroy(m_ref); GlobalVariable::remove(m_refName); GlobalVariable::remove(m_counterName); } @@ -117,9 +118,5 @@ public: } T ref() const {return m_ref;} - -protected: - virtual T create()=NULL; - virtual void destroy(T handle)=NULL; }; //+------------------------------------------------------------------+ diff --git a/Include/Zmq/ZmqMsg.mqh b/Include/Zmq/ZmqMsg.mqh index cd63a97..ebd0a21 100644 --- a/Include/Zmq/ZmqMsg.mqh +++ b/Include/Zmq/ZmqMsg.mqh @@ -49,12 +49,29 @@ protected: int get(int property) {return zmq_msg_get(this,property);} bool set(int property,int value) {return 0==zmq_msg_set(this,property,value);} intptr_t data() {return zmq_msg_data(this);} + bool setStringData(string data,bool nullterminated=false); public: ZmqMsg() {zmq_msg_init(this);} ZmqMsg(int size) {if(0!=zmq_msg_init_size(this,size)){Debug("Failed to init size msg: insufficient space");}} - ZmqMsg(string data,bool nullterminated=false); + ZmqMsg(string data,bool nullterminated=false) {setStringData(data,nullterminated);} ~ZmqMsg() {if(0!=zmq_msg_close(this)){Debug("Failed to close msg");}} + bool rebuild() + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return 0==zmq_msg_init(this); + } + bool rebuild(int size) + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return 0==zmq_msg_init_size(this,size); + } + bool rebuild(string data,bool nullterminated=false) + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return setStringData(data,nullterminated); + } + size_t size() {return zmq_msg_size(this);} void getData(uchar &data[]); @@ -71,13 +88,13 @@ public: //+------------------------------------------------------------------+ //| Initialize a utf-8 string message | //+------------------------------------------------------------------+ -ZmqMsg::ZmqMsg(string data,bool nullterminated) +bool ZmqMsg::setStringData(string data,bool nullterminated) { uchar array[]; StringToUtf8(data,array,nullterminated); - int size=ArraySize(array); - zmq_msg_init_size(this,size); - setData(array); + bool res=(0==zmq_msg_init_size(this,ArraySize(array))); + if(res)setData(array); + return res; } //+------------------------------------------------------------------+ //| Get message data as bytes array | diff --git a/README.md b/README.md index 4dca96c..e5ab17e 100644 --- a/README.md +++ b/README.md @@ -92,5 +92,7 @@ void OnStart() ## Changes +* 2017-06-08: Released 1.2: Fix GlobalHandle bug. Add rebuild method to ZmqMsg. + Complete all examples in ZMQ Guide Chanpter 1. * 2017-05-26: Released 1.1: add the ability to share a ZMQ context globally in a terminal * 2016-12-27: Released 1.0. diff --git a/Scripts/ZeroMQGuideExamples/Chapter1/TaskEvent.mq4 b/Scripts/ZeroMQGuideExamples/Chapter1/TaskEvent.mq4 new file mode 100644 index 0000000000000000000000000000000000000000..c91224e0eb60caeb27a9ce43462bdd46551c2fa7 GIT binary patch literal 4596 zcmd6q>u(!X5XI*+692<$D2d&YU~`^N*IbY|}oy0oSH|Y#-rG%ide|KV65`v8kQfH}=MQ zHnN#D?aW@==j{Lfn*Kbov7Ot6O>AI&J7KNN?%I}Z+m7AgewQ6Oyv^*;-ZEFeJ?k=a z#M6*ZAIl#8WTv&6ZuN|Hj<~7BRm=5&lw(%@nCL`iQ}$)i6>n~&MN-gB>vnY z2NOIz2gwPZU*M@cu4g89jO(NSUcPc?&oXa@j_oOa6<1D<*CD=oMR9imZwACUTCAFr zuO{rE z)ZjD6euMFbkIWYL4J^d^4s-Q3=4E!9`+Xlhu^0S~@T1`~TE50-@Nfqv?(nq9C*s^- zM$(U)#K)v(Nolf6r5O;tVjq)(%#l&tKiV^ovP0AzSUUsRBSwyy6?G?Mh4K@kSNzrN z3(4hIRnK;P$0Jvp=k}Ae?K|wI?9oLJd+eLJZ~E@y$30hO<*SQdLwp!Bd&c+|ekmK* zyQ==Q*;AR8`n_^oq>LH69P>S6L^`dvLQEVJZt_&j7YYOB=OXHav2&k)?E=({*IC%Z}2AF#pU!{#XkrXw0yHt}o zC@ha@7q&!HnPZ_&WX?M242imHGpZ9pI&=@T&aqM#BTrOes!O4LwI1bL##{9^p;%(A zaG!vf-bzJMhNOID6nC+D=aCf|{m?za3u&l2gV-gbF@>h zk<*Rk_A}2q+oZ>#+f^NEGBf7x)9<~s1N)vnGVb@N*#rB*_V|3u-3!LQrU%^f(Uhmi z^mFzWM&fY}|3Vuz(r52=YWPXEI`4<5f@0HYE#*`!RR-%UlA`9`I~3@tzvwUWruXIS z2U`l=F7jGuzhXH30Op5X=+uZc#MeNl%>C8p(^te(#W=c}ec}+NRSbl>I;X0I5Rmsx zY$79#nq8WoI+3Ixd)SFLI-{i0zj_WPJgI`?T#x?e6>r2zX^EoLoahk>OjCE%_P3z{ zPnkP|P3q=Oy|4F7}QpOMRo5nf4oS;)U1Rec-)Nq4?oVS@&b4>|N?D z1N~IT(UDd=u5EURt|N9_=4q^-u&S^2u+|gDIvaSsT~YX0WjWHD#9m*wd^&_}mB&pO z(&MC6O8_%(qqV?Q%PZTh^g@1VZ=>Mn`@T4Q}RglGExy_DOk#{E66=nRq) zFGolUUGS)0y=@rV<2onGz< +//+------------------------------------------------------------------+ +//| Task sink in MQL (adapted from C++ version) | +//| Binds PULL socket to tcp://localhost:5558 | +//| Collects results from workers via that socket | +//| | +//| Olivier Chamoux | +//+------------------------------------------------------------------+ +void OnStart() + { +//--- +// Prepare our context and socket + Context context; + Socket receiver(context,ZMQ_PULL); + receiver.bind("tcp://*:5558"); + +// Wait for start of batch + ZmqMsg message; + receiver.recv(message); + +// Start our clock now + uint tstart=GetTickCount(); +// Process 100 confirmations + string progress=""; + for(int i=0; i<100; i++) + { + receiver.recv(message); + if((i/10)*10==i) + progress+=":"; + else + progress+="."; + Comment(progress); + } +// Calculate and report duration of batch + uint tend=GetTickCount(); + Print(">>> Total elapsed time: ",tend-tstart," msec"); + } +//+------------------------------------------------------------------+ diff --git a/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 b/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 new file mode 100644 index 0000000..fa0a298 --- /dev/null +++ b/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 @@ -0,0 +1,52 @@ +//+------------------------------------------------------------------+ +//| TaskWorker.mq4 | +//| Copyright 2017, Bear Two Technologies Co., Ltd. | +//| dingmaotu@126.com | +//+------------------------------------------------------------------+ +#property copyright "Copyright 2017, Bear Two Technologies Co., Ltd." +#property link "dingmaotu@126.com" +#property version "1.00" +#property strict +#include +//+------------------------------------------------------------------+ +//| Task worker in MQL (adapted from C++ version) | +//| Connects PULL socket to tcp://localhost:5557 | +//| Collects workloads from ventilator via that socket | +//| Connects PUSH socket to tcp://localhost:5558 | +//| Sends results to sink via that socket | +//| | +//| Olivier Chamoux | +//+------------------------------------------------------------------+ +void OnStart() + { +//--- Share a single context in the terminal by the key "work" + Context context("work"); + +//--- Socket to receive messages on + Socket receiver(context,ZMQ_PULL); + receiver.connect("tcp://localhost:5557"); + +//--- Socket to send messages to + Socket sender(context,ZMQ_PUSH); + sender.connect("tcp://localhost:5558"); + +//--- Process tasks forever + string progress=""; + while(!IsStopped()) + { + ZmqMsg message; + receiver.recv(message); + //--- Workload in msecs + int workload=(int)StringToInteger(message.getData()); + //--- Do the work + Sleep(workload); + //--- Send results to sink + message.rebuild(); + sender.send(message); + + // Simple progress indicator for the viewer + progress+="."; + Comment(progress); + } + } +//+------------------------------------------------------------------+