3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-27 15:41:05 +08:00
Code Issues Projects Releases Wiki Activity

Revert "Add code to investigate data race"

Browse Source 
This reverts commit 5e0facda17b986943f0fe40a2dc6c98611655c35.
This commit is contained in:
Pieter Hintjens 2014-05-02 16:50:16 +02:00
parent d190325e4d
commit ded00177b3
3 changed files with 5 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/zmq.h
View File

@ -220,12 +220,6 @@ ZMQ_EXPORT int zmq_msg_get (zmq_msg_t *msg, int property);
ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int property, int optval);
ZMQ_EXPORT const char *zmq_msg_gets (zmq_msg_t *msg, const char *property);
// DAB - these are millisecond sleeps to bias data races
extern ZMQ_EXPORT int zmq_lb_race_window_1_size;
extern ZMQ_EXPORT int zmq_lb_race_window_2_size;
extern ZMQ_EXPORT int zmq_fq_race_window_1_size;
extern ZMQ_EXPORT int zmq_fq_race_window_2_size;
/******************************************************************************/
/* 0MQ socket definition. */

29
src/fq.cpp
View File

@ -16,36 +16,18 @@
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdio.h>
#include "fq.hpp"
#include "pipe.hpp"
#include "err.hpp"
#include "msg.hpp"
#ifdef ZMQ_HAVE_WINDOWS
# define msleep(milliseconds) {if(milliseconds) Sleep (milliseconds);}
#else
# include <unistd.h>
# define msleep(milliseconds) {if(milliseconds) usleep (static_cast <useconds_t> (milliseconds) * 1000);}
#endif
#define DB_TRACE(tag) int my_seq = ++seq; \
pthread_t self = pthread_self(); \
fprintf(stderr, "=> %12.12s thread=%lu this=%p seq=%d active=%lu\n", tag, self, (void *)this, my_seq, active) ; \
#define DB_TRACE_EXIT(tag) fprintf(stderr, "<= %12.12s thread=%lu this=%p seq=%d active=%lu\n", tag, self, (void *)this, my_seq, active) ; \
int zmq_fq_race_window_1_size = 0 ;
int zmq_fq_race_window_2_size = 0 ;
static int seq = 0 ;
zmq::fq_t::fq_t () :
active (0),
last_in (NULL),
current (0),
more (false)
{
DB_TRACE("fq_cons");
}
zmq::fq_t::~fq_t ()
@ -62,7 +44,6 @@ void zmq::fq_t::attach (pipe_t *pipe_)
void zmq::fq_t::pipe_terminated (pipe_t *pipe_)
{
DB_TRACE("fq_term") ;
const pipes_t::size_type index = pipes.index (pipe_);
// Remove the pipe from the list; adjust number of active pipes
@ -79,7 +60,6 @@ void zmq::fq_t::pipe_terminated (pipe_t *pipe_)
saved_credential = last_in->get_credential ();
last_in = NULL;
}
DB_TRACE_EXIT("fq_term") ;
}
void zmq::fq_t::activated (pipe_t *pipe_)
@ -96,22 +76,17 @@ int zmq::fq_t::recv (msg_t *msg_)
int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
{
DB_TRACE("fq_recvpipe");
// Deallocate old content of the message.
int rc = msg_->close ();
errno_assert (rc == 0);
// Round-robin over the pipes to get the next message.
while (active > 0) {
// DAB - bias the race to provoke problems with read
msleep(zmq_fq_race_window_1_size) ;
// Try to fetch new message. If we've already read part of the message
// subsequent part should be immediately available.
bool fetched = pipes [current]->read (msg_);
// DAB - bias the race to provoke problems with %
msleep(zmq_fq_race_window_2_size) ;
// Note that when message is not fetched, current pipe is deactivated
// and replaced by another active pipe. Thus we don't have to increase
// the 'current' pointer.
@ -123,7 +98,6 @@ int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
last_in = pipes [current];
current = (current + 1) % active;
}
DB_TRACE_EXIT("fq_recvpipe");
return 0;
}
@ -143,7 +117,6 @@ int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
rc = msg_->init ();
errno_assert (rc == 0);
errno = EAGAIN;
DB_TRACE_EXIT("fq_recvpipe");
return -1;
}

42
src/lb.cpp
View File

@ -16,29 +16,18 @@
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdio.h>
#include "lb.hpp"
#include "pipe.hpp"
#include "err.hpp"
#include "msg.hpp"
#define DB_TRACE(tag) int my_seq = ++seq; \
pthread_t self = pthread_self(); \
fprintf(stderr, "=> %12.12s thread=%lu this=%p seq=%d active=%lu\n", tag, self, (void *)this, my_seq, active) ; \
#define DB_TRACE_EXIT(tag) fprintf(stderr, "<= %12.12s thread=%lu this=%p seq=%d active=%lu\n", tag, self, (void *)this, my_seq, active) ; \
int zmq_lb_race_window_1_size = 0 ;
int zmq_lb_race_window_2_size = 0 ;
static int seq = 0 ;
zmq::lb_t::lb_t () :
active (0),
active (0),
current (0),
more (false),
dropping (false)
{
DB_TRACE("lb_cons") ;
}
zmq::lb_t::~lb_t ()
@ -54,7 +43,6 @@ void zmq::lb_t::attach (pipe_t *pipe_)
void zmq::lb_t::pipe_terminated (pipe_t *pipe_)
{
DB_TRACE("lb_term") ;
pipes_t::size_type index = pipes.index (pipe_);
// If we are in the middle of multipart message and current pipe
@ -71,7 +59,6 @@ void zmq::lb_t::pipe_terminated (pipe_t *pipe_)
current = 0;
}
pipes.erase (pipe_);
DB_TRACE_EXIT("lb_term") ;
}
void zmq::lb_t::activated (pipe_t *pipe_)
@ -86,16 +73,8 @@ int zmq::lb_t::send (msg_t *msg_)
return sendpipe (msg_, NULL);
}
#ifdef ZMQ_HAVE_WINDOWS
# define msleep(milliseconds) Sleep (milliseconds);
#else
# include <unistd.h>
# define msleep(milliseconds) usleep (static_cast <useconds_t> (milliseconds) * 1000);
#endif
int zmq::lb_t::sendpipe (msg_t *msg_, pipe_t **pipe_)
{
DB_TRACE("lb_sendpipe") ;
// Drop the message if required. If we are at the end of the message
// switch back to non-dropping mode.
if (dropping) {
@ -111,20 +90,13 @@ int zmq::lb_t::sendpipe (msg_t *msg_, pipe_t **pipe_)
}
while (active > 0) {
// DAB - bias the race to provoke problems with write
msleep(zmq_lb_race_window_1_size ) ;
if (pipes [current]->write (msg_))
if (pipes [current]->write (msg_))
{
if (pipe_)
*pipe_ = pipes [current];
break;
}
if (!(!more))
{
DB_TRACE("lb_assert");
fflush(stderr) ;
}
zmq_assert (!more);
active--;
if (current < active)
@ -136,17 +108,12 @@ int zmq::lb_t::sendpipe (msg_t *msg_, pipe_t **pipe_)
// If there are no pipes we cannot send the message.
if (active == 0) {
errno = EAGAIN;
DB_TRACE_EXIT("lb_sendpipe") ;
return -1;
}
// If it's final part of the message we can flush it downstream and
// continue round-robining (load balance).
more = msg_->flags () & msg_t::more? true: false;
// DAB - bias the race to provoke problems with %
msleep(zmq_lb_race_window_2_size) ;
if (!more) {
pipes [current]->flush ();
current = (current + 1) % active;
@ -156,7 +123,6 @@ int zmq::lb_t::sendpipe (msg_t *msg_, pipe_t **pipe_)
int rc = msg_->init ();
errno_assert (rc == 0);
DB_TRACE_EXIT("lb_sendpipe") ;
return 0;
}