GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — Merge pull request #101 from mujin/revert-97-silence_zmq_bullseye · mujin/controllerclientcpp@42f4249

File tree

Expand file treeCollapse file tree

Open diff view settings

Filter options

include/mujincontrollerclient
- mujinjson.h
src

Expand file treeCollapse file tree

Open diff view settings

Collapse file

`‎include/mujincontrollerclient/mujinjson.h‎`

Copy file name to clipboardExpand all lines: include/mujincontrollerclient/mujinjson.h

+1-1Lines changed: 1 addition & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -164,7 +164,7 @@ inline void ParseJson(rapidjson::Document& d, const std::string& str) {`
`164`	`164`	`inline void ParseJson(rapidjson::Document& d, std::istream& is) {`
`165`	`165`	`rapidjson::IStreamWrapper isw(is);`
`166`	`166`	`// see note in: void ParseJson(rapidjson::Document& d, const std::string& str)`
`167`		`- rapidjson::Document tempDoc;`
	`167`	`+ rapidjson::Document(tempDoc);`
`168`	`168`	`tempDoc.ParseStream<rapidjson::kParseFullPrecisionFlag>(isw); // parse float in full precision mode`
`169`	`169`	`if (tempDoc.HasParseError()) {`
`170`	`170`	`throw MujinJSONException(boost::str(boost::format("Json stream is invalid (offset %u) %s")%((unsigned)d.GetErrorOffset())%GetParseError_En(d.GetParseError())), MJE_Failed);`

Collapse file

`‎src/binpickingtask.cpp‎`

Copy file name to clipboardExpand all lines: src/binpickingtask.cpp

+27-20Lines changed: 27 additions & 20 deletions

Original file line number	Diff line number	Diff line change
`@@ -1785,12 +1785,14 @@ void BinPickingTaskResource::_HeartbeatMonitorThread(const double reinitializeti`
`1785`	`1785`	`socket.reset();`
`1786`	`1786`	`}`
`1787`	`1787`	`socket.reset(new zmq::socket_t((*_zmqcontext.get()),ZMQ_SUB));`
`1788`		`- socket->set(zmq::sockopt::tcp_keepalive, 1); // turn on tcp keepalive, do these configuration before connect`
`1789`		`- socket->set(zmq::sockopt::tcp_keepalive_idle, 2); // the interval between the last data packet sent (simple ACKs are not considered data) and the first keepalive probe; after the connection is marked to need keepalive, this counter is not used any further`
`1790`		`- socket->set(zmq::sockopt::tcp_keepalive_intvl, 2); // the interval between subsequential keepalive probes, regardless of what the connection has exchanged in the meantime`
`1791`		`- socket->set(zmq::sockopt::tcp_keepalive_cnt, 2); // the number of unacknowledged probes to send before considering the connection dead and notifying the application layer`
`1792`		`- socket->connect("tcp://" + _mujinControllerIp + ":" + std::to_string(_heartbeatPort));`
`1793`		`- socket->set(zmq::sockopt::subscribe, "");`
	`1788`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE, 1); // turn on tcp keepalive, do these configuration before connect`
	`1789`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_IDLE, 2); // the interval between the last data packet sent (simple ACKs are not considered data) and the first keepalive probe; after the connection is marked to need keepalive, this counter is not used any further`
	`1790`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_INTVL, 2); // the interval between subsequential keepalive probes, regardless of what the connection has exchanged in the meantime`
	`1791`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_CNT, 2); // the number of unacknowledged probes to send before considering the connection dead and notifying the application layer`
	`1792`	`+ std::stringstream ss; ss << std::setprecision(std::numeric_limits<double>::digits10+1);`
	`1793`	`+ ss << _heartbeatPort;`
	`1794`	`+ socket->connect(("tcp://"+ _mujinControllerIp+":"+ss.str()).c_str());`
	`1795`	`+ socket->setsockopt(ZMQ_SUBSCRIBE, "", 0);`
`1794`	`1796`
`1795`	`1797`	`zmq::pollitem_t pollitem;`
`1796`	`1798`	`memset(&pollitem, 0, sizeof(zmq::pollitem_t));`
`@@ -1799,12 +1801,13 @@ void BinPickingTaskResource::_HeartbeatMonitorThread(const double reinitializeti`
`1799`	`1801`
`1800`	`1802`	`unsigned long long lastheartbeat = GetMilliTime();`
`1801`	`1803`	`while (!_bShutdownHeartbeatMonitor && (GetMilliTime() - lastheartbeat) / 1000.0f < reinitializetimeout) {`
`1802`		`- zmq::poll(&pollitem, 1, std::chrono::milliseconds{50}); // wait 50 ms for message`
	`1804`	`+ zmq::poll(&pollitem,1, 50); // wait 50 ms for message`
`1803`	`1805`	`if (pollitem.revents & ZMQ_POLLIN) {`
`1804`	`1806`	`zmq::message_t reply;`
`1805`		`- socket->recv(reply);`
	`1807`	`+ socket->recv(&reply);`
	`1808`	`+ std::string replystring((char *)reply.data (), (size_t)reply.size());`
`1806`	`1809`	`//if ((size_t)reply.size() == 1 && ((char *)reply.data())[0]==255) {`
`1807`		`- if (reply.to_string() == "255") {`
	`1810`	`+ if (replystring == "255") {`
`1808`	`1811`	`lastheartbeat = GetMilliTime();`
`1809`	`1812`	`}`
`1810`	`1813`	`}`
`@@ -1824,36 +1827,37 @@ void BinPickingTaskResource::_HeartbeatMonitorThread(const double reinitializeti`
`1824`	`1827`	`std::string utils::GetHeartbeat(const std::string& endpoint) {`
`1825`	`1828`	`zmq::context_t zmqcontext(1);`
`1826`	`1829`	`zmq::socket_t socket(zmqcontext, ZMQ_SUB);`
`1827`		`- socket.connect(endpoint);`
`1828`		`- socket.set(zmq::sockopt::subscribe, "");`
	`1830`	`+ socket.connect(endpoint.c_str());`
	`1831`	`+ socket.setsockopt(ZMQ_SUBSCRIBE, "", 0);`
`1829`	`1832`
`1830`	`1833`	`zmq::pollitem_t pollitem;`
`1831`	`1834`	`memset(&pollitem, 0, sizeof(zmq::pollitem_t));`
`1832`	`1835`	`pollitem.socket = socket;`
`1833`	`1836`	`pollitem.events = ZMQ_POLLIN;`
`1834`	`1837`
`1835`		`- zmq::poll(&pollitem, 1, std::chrono::milliseconds{50}); // wait 50 ms for message`
	`1838`	`+ zmq::poll(&pollitem,1, 50); // wait 50 ms for message`
`1836`	`1839`	`if (!(pollitem.revents & ZMQ_POLLIN)) {`
`1837`	`1840`	`return "";`
`1838`	`1841`	`}`
`1839`	`1842`
`1840`	`1843`	`zmq::message_t reply;`
`1841`		`- socket.recv(reply);`
	`1844`	`+ socket.recv(&reply);`
	`1845`	`+ const std::string received((char *)reply.data (), (size_t)reply.size());`
`1842`	`1846`	`#ifndef _WIN32`
`1843`		`- return reply.to_string();`
	`1847`	`+ return received;`
`1844`	`1848`	`#else`
`1845`	`1849`	`// sometimes buffer can container \n or \\, which windows does not like`
`1846`	`1850`	`std::string newbuffer;`
`1847`	`1851`	`std::vector< std::pair<std::string, std::string> > serachpairs(2);`
`1848`	`1852`	`serachpairs[0].first = "\n"; serachpairs[0].second = "";`
`1849`	`1853`	`serachpairs[1].first = "\\"; serachpairs[1].second = "";`
`1850`		`- SearchAndReplace(newbuffer, reply.to_string(), serachpairs);`
	`1854`	`+ SearchAndReplace(newbuffer, received, serachpairs);`
`1851`	`1855`	`return newbuffer;`
`1852`	`1856`	`#endif`
`1853`	`1857`	`}`
`1854`	`1858`
`1855`		`-namespace {`
`1856`	`1859`
	`1860`	`+namespace {`
`1857`	`1861`	`std::string FindSmallestSlaveRequestId(const rapidjson::Value& pt) {`
`1858`	`1862`	`// get all slave request ids`
`1859`	`1863`	`std::vector<std::string> slavereqids;`
`@@ -1894,11 +1898,13 @@ std::string FindSmallestSlaveRequestId(const rapidjson::Value& pt) {`
`1894`	`1898`	`return slavereqids[smallest_suffix_index];`
`1895`	`1899`	`}`
`1896`	`1900`
`1897`		`-std::string GetValueForSmallestSlaveRequestId(const std::string& heartbeat, const std::string& key)`
	`1901`	`+std::string GetValueForSmallestSlaveRequestId(const std::string& heartbeat,`
	`1902`	`+ const std::string& key)`
`1898`	`1903`	`{`
`1899`	`1904`
`1900`	`1905`	`rapidjson::Document pt(rapidjson::kObjectType);`
`1901`		`- ParseJson(pt, heartbeat);`
	`1906`	`+ std::stringstream ss(heartbeat);`
	`1907`	`+ ParseJson(pt, ss.str());`
`1902`	`1908`	`try {`
`1903`	`1909`	`const std::string slavereqid = FindSmallestSlaveRequestId(pt);`
`1904`	`1910`	`std::string result;`
`@@ -1910,8 +1916,8 @@ std::string GetValueForSmallestSlaveRequestId(const std::string& heartbeat, cons`
`1910`	`1916`	`}`
`1911`	`1917`
`1912`	`1918`	`}`
	`1919`	`+}`
`1913`	`1920`
`1914`		`-} // anonymous namespace`
`1915`	`1921`
`1916`	`1922`	`std::string mujinclient::utils::GetScenePkFromHeartbeat(const std::string& heartbeat) {`
`1917`	`1923`	`static const std::string prefix("mujin:/");`
`@@ -1920,7 +1926,8 @@ std::string mujinclient::utils::GetScenePkFromHeartbeat(const std::string& heart`
`1920`	`1926`
`1921`	`1927`	`std::string utils::GetSlaveRequestIdFromHeartbeat(const std::string& heartbeat) {`
`1922`	`1928`	`rapidjson::Document pt;`
`1923`		`- ParseJson(pt, heartbeat);`
	`1929`	`+ std::stringstream ss(heartbeat);`
	`1930`	`+ ParseJson(pt, ss.str());`
`1924`	`1931`	`try {`
`1925`	`1932`	`static const std::string prefix("slaverequestid-");`
`1926`	`1933`	`return FindSmallestSlaveRequestId(pt).substr(prefix.length());`

Collapse file

`‎src/binpickingtaskzmq.cpp‎`

Copy file name to clipboardExpand all lines: src/binpickingtaskzmq.cpp

+16-10Lines changed: 16 additions & 10 deletions

Original file line number	Diff line number	Diff line change
`@@ -267,26 +267,30 @@ void BinPickingTaskZmqResource::_HeartbeatMonitorThread(const double reinitializ`
`267`	`267`	`socket.reset();`
`268`	`268`	`}`
`269`	`269`	`socket.reset(new zmq::socket_t((*_zmqcontext.get()),ZMQ_SUB));`
`270`		`- socket->set(zmq::sockopt::tcp_keepalive, 1); // turn on tcp keepalive, do these configuration before connect`
`271`		`- socket->set(zmq::sockopt::tcp_keepalive_idle, 2); // the interval between the last data packet sent (simple ACKs are not considered data) and the first keepalive probe; after the connection is marked to need keepalive, this counter is not used any further`
`272`		`- socket->set(zmq::sockopt::tcp_keepalive_intvl, 2); // the interval between subsequential keepalive probes, regardless of what the connection has exchanged in the meantime`
`273`		`- socket->set(zmq::sockopt::tcp_keepalive_cnt, 2); // the number of unacknowledged probes to send before considering the connection dead and notifying the application layer`
`274`		`- socket->connect("tcp://" + _mujinControllerIp + ":" + std::to_string(_heartbeatPort));`
`275`		`- socket->set(zmq::sockopt::subscribe, "");`
	`270`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE, 1); // turn on tcp keepalive, do these configuration before connect`
	`271`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_IDLE, 2); // the interval between the last data packet sent (simple ACKs are not considered data) and the first keepalive probe; after the connection is marked to need keepalive, this counter is not used any further`
	`272`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_INTVL, 2); // the interval between subsequential keepalive probes, regardless of what the connection has exchanged in the meantime`
	`273`	`+ socket->setsockopt(ZMQ_TCP_KEEPALIVE_CNT, 2); // the number of unacknowledged probes to send before considering the connection dead and notifying the application layer`
	`274`	`+ std::stringstream ss; ss << std::setprecision(std::numeric_limits<double>::digits10+1);`
	`275`	`+ ss << _heartbeatPort;`
	`276`	`+ socket->connect (("tcp://"+ _mujinControllerIp+":"+ss.str()).c_str());`
	`277`	`+ socket->setsockopt(ZMQ_SUBSCRIBE, "", 0);`
`276`	`278`
`277`	`279`	`zmq::pollitem_t pollitem;`
`278`	`280`	`memset(&pollitem, 0, sizeof(zmq::pollitem_t));`
`279`	`281`	`pollitem.socket = socket->operator void*();`
`280`	`282`	`pollitem.events = ZMQ_POLLIN;`
`281`	`283`	`unsigned long long lastheartbeat = GetMilliTime();`
`282`	`284`	`while (!_bShutdownHeartbeatMonitor && (GetMilliTime() - lastheartbeat) / 1000.0f < reinitializetimeout) {`
`283`		`- zmq::poll(&pollitem, 1, std::chrono::milliseconds{50}); // wait 50 ms for message`
	`285`	`+ zmq::poll(&pollitem,1, 50); // wait 50 ms for message`
`284`	`286`	`if (pollitem.revents & ZMQ_POLLIN) {`
`285`	`287`	`zmq::message_t reply;`
`286`		`- socket->recv(reply);`
	`288`	`+ socket->recv(&reply);`
	`289`	`+ std::string replystring((char *)reply.data (), (size_t)reply.size());`
`287`	`290`	`rapidjson::Document pt(rapidjson::kObjectType);`
`288`	`291`	`try{`
`289`		`- ParseJson(pt, reply.to_string());`
	`292`	`+ std::stringstream replystring_ss(replystring);`
	`293`	`+ ParseJson(pt, replystring_ss.str());`
`290`	`294`	`heartbeat.Parse(pt);`
`291`	`295`	`{`
`292`	`296`	`boost::mutex::scoped_lock lock(_mutexTaskState);`
`@@ -300,7 +304,7 @@ void BinPickingTaskZmqResource::_HeartbeatMonitorThread(const double reinitializ`
`300`	`304`	`}`
`301`	`305`	`catch (std::exception const &e) {`
`302`	`306`	`MUJIN_LOG_ERROR("HeartBeat reply is not JSON");`
`303`		`- MUJIN_LOG_ERROR(reply.to_string());`
	`307`	`+ MUJIN_LOG_ERROR(replystring);`
`304`	`308`	`MUJIN_LOG_ERROR(e.what());`
`305`	`309`	`continue;`
`306`	`310`	`}`
`@@ -315,4 +319,6 @@ void BinPickingTaskZmqResource::_HeartbeatMonitorThread(const double reinitializ`
`315`	`319`	`MUJIN_LOG_DEBUG(str(boost::format("Stopped controller %s monitoring thread on port %d for slaverequestid=%s.")%_mujinControllerIp%_heartbeatPort%_slaverequestid));`
`316`	`320`	`}`
`317`	`321`
	`322`	`+`
	`323`	`+`
`318`	`324`	`} // end namespace mujinclient`

0 commit comments

Comments

0 (0)

You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.