Skip to content
Projects
Groups
Snippets
Help
Loading...
Sign in / Register
Toggle navigation
L
libzmq
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Packages
Packages
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
submodule
libzmq
Commits
f8f79137
Commit
f8f79137
authored
Dec 24, 2019
by
Simon Giesecke
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Problem: irregular loop conditions
Solution: use standard loop constructs and optimize some loops
parent
628adf1c
Hide whitespace changes
Inline
Side-by-side
Showing
10 changed files
with
72 additions
and
57 deletions
+72
-57
ctx.cpp
src/ctx.cpp
+15
-8
generic_mtrie_impl.hpp
src/generic_mtrie_impl.hpp
+15
-19
ipc_listener.cpp
src/ipc_listener.cpp
+4
-3
pipe.cpp
src/pipe.cpp
+4
-3
radix_tree.cpp
src/radix_tree.cpp
+18
-14
socket_base.cpp
src/socket_base.cpp
+4
-3
socket_base.hpp
src/socket_base.hpp
+3
-2
tcp_listener.cpp
src/tcp_listener.cpp
+4
-2
thread.cpp
src/thread.cpp
+3
-2
ws_encoder.cpp
src/ws_encoder.cpp
+2
-1
No files found.
src/ctx.cpp
View file @
f8f79137
...
...
@@ -117,12 +117,13 @@ zmq::ctx_t::~ctx_t ()
// Ask I/O threads to terminate. If stop signal wasn't sent to I/O
// thread subsequent invocation of destructor would hang-up.
for
(
io_threads_t
::
size_type
i
=
0
;
i
!=
_io_threads
.
size
();
i
++
)
{
const
io_threads_t
::
size_type
io_threads_size
=
_io_threads
.
size
();
for
(
io_threads_t
::
size_type
i
=
0
;
i
!=
io_threads_size
;
i
++
)
{
_io_threads
[
i
]
->
stop
();
}
// Wait till I/O threads actually terminate.
for
(
io_threads_t
::
size_type
i
=
0
;
i
!=
_io_threads
.
size
()
;
i
++
)
{
for
(
io_threads_t
::
size_type
i
=
0
;
i
!=
io_threads_size
;
i
++
)
{
LIBZMQ_DELETE
(
_io_threads
[
i
]);
}
...
...
@@ -176,9 +177,10 @@ int zmq::ctx_t::terminate ()
if
(
_pid
!=
getpid
())
{
// we are a forked child process. Close all file descriptors
// inherited from the parent.
for
(
sockets_t
::
size_type
i
=
0
;
i
!=
_sockets
.
size
();
i
++
)
for
(
sockets_t
::
size_type
i
=
0
,
size
=
_sockets
.
size
();
i
!=
size
;
i
++
)
{
_sockets
[
i
]
->
get_mailbox
()
->
forked
();
}
_term_mailbox
.
forked
();
}
#endif
...
...
@@ -193,8 +195,10 @@ int zmq::ctx_t::terminate ()
// First send stop command to sockets so that any blocking calls
// can be interrupted. If there are no sockets we can ask reaper
// thread to stop.
for
(
sockets_t
::
size_type
i
=
0
;
i
!=
_sockets
.
size
();
i
++
)
for
(
sockets_t
::
size_type
i
=
0
,
size
=
_sockets
.
size
();
i
!=
size
;
i
++
)
{
_sockets
[
i
]
->
stop
();
}
if
(
_sockets
.
empty
())
_reaper
->
stop
();
}
...
...
@@ -239,8 +243,10 @@ int zmq::ctx_t::shutdown ()
// Send stop command to sockets so that any blocking calls
// can be interrupted. If there are no sockets we can ask reaper
// thread to stop.
for
(
sockets_t
::
size_type
i
=
0
;
i
!=
_sockets
.
size
();
i
++
)
for
(
sockets_t
::
size_type
i
=
0
,
size
=
_sockets
.
size
();
i
!=
size
;
i
++
)
{
_sockets
[
i
]
->
stop
();
}
if
(
_sockets
.
empty
())
_reaper
->
stop
();
}
...
...
@@ -666,9 +672,10 @@ zmq::io_thread_t *zmq::ctx_t::choose_io_thread (uint64_t affinity_)
// Find the I/O thread with minimum load.
int
min_load
=
-
1
;
io_thread_t
*
selected_io_thread
=
NULL
;
for
(
io_threads_t
::
size_type
i
=
0
;
i
!=
_io_threads
.
size
();
i
++
)
{
for
(
io_threads_t
::
size_type
i
=
0
,
size
=
_io_threads
.
size
();
i
!=
size
;
i
++
)
{
if
(
!
affinity_
||
(
affinity_
&
(
uint64_t
(
1
)
<<
i
)))
{
int
load
=
_io_threads
[
i
]
->
get_load
();
const
int
load
=
_io_threads
[
i
]
->
get_load
();
if
(
selected_io_thread
==
NULL
||
load
<
min_load
)
{
min_load
=
load
;
selected_io_thread
=
_io_threads
[
i
];
...
...
src/generic_mtrie_impl.hpp
View file @
f8f79137
...
...
@@ -427,13 +427,14 @@ void zmq::generic_mtrie_t<T>::match (prefix_t data_,
void
(
*
func_
)
(
value_t
*
pipe_
,
Arg
arg_
),
Arg
arg_
)
{
generic_mtrie_t
*
current
=
this
;
while
(
true
)
{
for
(
generic_mtrie_t
*
current
=
this
;
current
;
data_
++
,
size_
--
)
{
// Signal the pipes attached to this node.
if
(
current
->
_pipes
)
{
for
(
typename
pipes_t
::
iterator
it
=
current
->
_pipes
->
begin
();
it
!=
current
->
_pipes
->
end
();
++
it
)
for
(
typename
pipes_t
::
iterator
it
=
current
->
_pipes
->
begin
(),
end
=
current
->
_pipes
->
end
();
it
!=
end
;
++
it
)
{
func_
(
*
it
,
arg_
);
}
}
// If we are at the end of the message, there's nothing more to match.
...
...
@@ -444,25 +445,20 @@ void zmq::generic_mtrie_t<T>::match (prefix_t data_,
if
(
current
->
_count
==
0
)
break
;
// If there's one subnode (optimisation).
if
(
current
->
_count
==
1
)
{
if
(
data_
[
0
]
!=
current
->
_min
)
// If there's one subnode (optimisation).
if
(
data_
[
0
]
!=
current
->
_min
)
{
break
;
}
current
=
current
->
_next
.
node
;
data_
++
;
size_
--
;
continue
;
}
else
{
// If there are multiple subnodes.
if
(
data_
[
0
]
<
current
->
_min
||
data_
[
0
]
>=
current
->
_min
+
current
->
_count
)
{
break
;
}
current
=
current
->
_next
.
table
[
data_
[
0
]
-
current
->
_min
];
}
// If there are multiple subnodes.
if
(
data_
[
0
]
<
current
->
_min
||
data_
[
0
]
>=
current
->
_min
+
current
->
_count
)
break
;
if
(
!
current
->
_next
.
table
[
data_
[
0
]
-
current
->
_min
])
break
;
current
=
current
->
_next
.
table
[
data_
[
0
]
-
current
->
_min
];
data_
++
;
size_
--
;
}
}
...
...
src/ipc_listener.cpp
View file @
f8f79137
...
...
@@ -253,9 +253,10 @@ bool zmq::ipc_listener_t::filter (fd_t sock_)
if
(
!
(
pw
=
getpwuid
(
cred
.
uid
)))
return
false
;
for
(
options_t
::
ipc_gid_accept_filters_t
::
const_iterator
it
=
options
.
ipc_gid_accept_filters
.
begin
();
it
!=
options
.
ipc_gid_accept_filters
.
end
();
it
++
)
{
for
(
options_t
::
ipc_gid_accept_filters_t
::
const_iterator
it
=
options
.
ipc_gid_accept_filters
.
begin
(),
end
=
options
.
ipc_gid_accept_filters
.
end
();
it
!=
end
;
it
++
)
{
if
(
!
(
gr
=
getgrgid
(
*
it
)))
continue
;
for
(
char
**
mem
=
gr
->
gr_mem
;
*
mem
;
mem
++
)
{
...
...
src/pipe.cpp
View file @
f8f79137
...
...
@@ -188,7 +188,7 @@ bool zmq::pipe_t::read (msg_t *msg_)
if
(
unlikely
(
_state
!=
active
&&
_state
!=
waiting_for_delimiter
))
return
false
;
for
(
bool
payload_read
=
false
;
!
payload_read
;
)
{
while
(
true
)
{
if
(
!
_in_pipe
->
read
(
msg_
))
{
_in_active
=
false
;
return
false
;
...
...
@@ -198,8 +198,9 @@ bool zmq::pipe_t::read (msg_t *msg_)
if
(
unlikely
(
msg_
->
is_credential
()))
{
const
int
rc
=
msg_
->
close
();
zmq_assert
(
rc
==
0
);
}
else
payload_read
=
true
;
}
else
{
break
;
}
}
// If delimiter was read, start termination process of the pipe.
...
...
src/radix_tree.cpp
View file @
f8f79137
...
...
@@ -34,6 +34,7 @@
#include <stdlib.h>
#include <string.h>
#include <iterator>
#include <vector>
node_t
::
node_t
(
unsigned
char
*
data_
)
:
_data
(
data_
)
...
...
@@ -187,7 +188,7 @@ zmq::radix_tree_t::radix_tree_t () : _root (make_node (0, 0, 0)), _size (0)
static
void
free_nodes
(
node_t
node_
)
{
for
(
size_t
i
=
0
;
i
<
node_
.
edgecount
()
;
++
i
)
for
(
size_t
i
=
0
,
count
=
node_
.
edgecount
();
i
<
count
;
++
i
)
free_nodes
(
node_
.
node_at
(
i
));
free
(
node_
.
_data
);
}
...
...
@@ -234,18 +235,19 @@ match_result_t zmq::radix_tree_t::match (const unsigned char *key_,
size_t
parent_edge_index
=
0
;
while
(
current_node
.
prefix_length
()
>
0
||
current_node
.
edgecount
()
>
0
)
{
const
unsigned
char
*
const
prefix
=
current_node
.
prefix
();
const
size_t
prefix_length
=
current_node
.
prefix_length
();
for
(
prefix_byte_index
=
0
;
prefix_byte_index
<
current_node
.
prefix_length
()
&&
key_byte_index
<
key_size_
;
prefix_byte_index
<
prefix_length
&&
key_byte_index
<
key_size_
;
++
prefix_byte_index
,
++
key_byte_index
)
{
if
(
current_node
.
prefix
()[
prefix_byte_index
]
!=
key_
[
key_byte_index
])
if
(
prefix
[
prefix_byte_index
]
!=
key_
[
key_byte_index
])
break
;
}
// Even if a prefix of the key matches and we're doing a
// lookup, this means we've found a matching subscription.
if
(
is_lookup_
&&
prefix_byte_index
==
current_node
.
prefix_length
()
if
(
is_lookup_
&&
prefix_byte_index
==
prefix_length
&&
current_node
.
refcount
()
>
0
)
{
key_byte_index
=
key_size_
;
break
;
...
...
@@ -253,14 +255,14 @@ match_result_t zmq::radix_tree_t::match (const unsigned char *key_,
// There was a mismatch or we've matched the whole key, so
// there's nothing more to do.
if
(
prefix_byte_index
!=
current_node
.
prefix_length
()
||
key_byte_index
==
key_size_
)
if
(
prefix_byte_index
!=
prefix_length
||
key_byte_index
==
key_size_
)
break
;
// We need to match the rest of the key. Check if there's an
// outgoing edge from this node.
node_t
next_node
=
current_node
;
for
(
size_t
i
=
0
;
i
<
current_node
.
edgecount
();
++
i
)
{
for
(
size_t
i
=
0
,
edgecount
=
current_node
.
edgecount
();
i
<
edgecount
;
++
i
)
{
if
(
current_node
.
first_byte_at
(
i
)
==
key_
[
key_byte_index
])
{
parent_edge_index
=
edge_index
;
edge_index
=
i
;
...
...
@@ -543,18 +545,20 @@ visit_keys (node_t node_,
void
(
*
func_
)
(
unsigned
char
*
data_
,
size_t
size_
,
void
*
arg_
),
void
*
arg_
)
{
for
(
size_t
i
=
0
;
i
<
node_
.
prefix_length
();
++
i
)
buffer_
.
push_back
(
node_
.
prefix
()[
i
]);
const
size_t
prefix_length
=
node_
.
prefix_length
();
buffer_
.
reserve
(
buffer_
.
size
()
+
prefix_length
);
std
::
copy
(
node_
.
prefix
(),
node_
.
prefix
()
+
prefix_length
,
std
::
back_inserter
(
buffer_
));
if
(
node_
.
refcount
()
>
0
)
{
zmq_assert
(
!
buffer_
.
empty
());
func_
(
&
buffer_
[
0
],
buffer_
.
size
(),
arg_
);
}
for
(
size_t
i
=
0
;
i
<
node_
.
edgecount
();
++
i
)
for
(
size_t
i
=
0
,
edgecount
=
node_
.
edgecount
();
i
<
edgecount
;
++
i
)
{
visit_keys
(
node_
.
node_at
(
i
),
buffer_
,
func_
,
arg_
);
for
(
size_t
i
=
0
;
i
<
node_
.
prefix_length
();
++
i
)
buffer_
.
pop_back
(
);
}
buffer_
.
resize
(
buffer_
.
size
()
-
prefix_length
);
}
void
zmq
::
radix_tree_t
::
apply
(
...
...
src/socket_base.cpp
View file @
f8f79137
...
...
@@ -1469,8 +1469,9 @@ void zmq::socket_base_t::process_term (int linger_)
unregister_endpoints
(
this
);
// Ask all attached pipes to terminate.
for
(
pipes_t
::
size_type
i
=
0
;
i
!=
_pipes
.
size
();
++
i
)
for
(
pipes_t
::
size_type
i
=
0
,
size
=
_pipes
.
size
();
i
!=
size
;
++
i
)
{
_pipes
[
i
]
->
terminate
(
false
);
}
register_term_acks
(
static_cast
<
int
>
(
_pipes
.
size
()));
// Continue the termination process immediately.
...
...
@@ -1515,7 +1516,7 @@ int zmq::socket_base_t::query_pipes_stats ()
errno
=
EAGAIN
;
return
-
1
;
}
for
(
pipes_t
::
size_type
i
=
0
;
i
!=
_pipes
.
size
()
;
++
i
)
{
for
(
pipes_t
::
size_type
i
=
0
,
size
=
_pipes
.
size
();
i
!=
size
;
++
i
)
{
_pipes
[
i
]
->
send_stats_to_peer
(
this
);
}
...
...
@@ -1525,7 +1526,7 @@ int zmq::socket_base_t::query_pipes_stats ()
void
zmq
::
socket_base_t
::
update_pipe_options
(
int
option_
)
{
if
(
option_
==
ZMQ_SNDHWM
||
option_
==
ZMQ_RCVHWM
)
{
for
(
pipes_t
::
size_type
i
=
0
;
i
!=
_pipes
.
size
()
;
++
i
)
{
for
(
pipes_t
::
size_type
i
=
0
,
size
=
_pipes
.
size
();
i
!=
size
;
++
i
)
{
_pipes
[
i
]
->
set_hwms
(
options
.
rcvhwm
,
options
.
sndhwm
);
_pipes
[
i
]
->
send_hwms_to_peer
(
options
.
sndhwm
,
options
.
rcvhwm
);
}
...
...
src/socket_base.hpp
View file @
f8f79137
...
...
@@ -376,8 +376,9 @@ class routing_socket_base_t : public socket_base_t
template
<
typename
Func
>
bool
any_of_out_pipes
(
Func
func_
)
{
bool
res
=
false
;
for
(
out_pipes_t
::
iterator
it
=
_out_pipes
.
begin
();
it
!=
_out_pipes
.
end
()
&&
!
res
;
++
it
)
{
for
(
out_pipes_t
::
iterator
it
=
_out_pipes
.
begin
(),
end
=
_out_pipes
.
end
();
it
!=
end
&&
!
res
;
++
it
)
{
res
|=
func_
(
*
it
->
second
.
pipe
);
}
...
...
src/tcp_listener.cpp
View file @
f8f79137
...
...
@@ -232,8 +232,10 @@ zmq::fd_t zmq::tcp_listener_t::accept ()
if
(
!
options
.
tcp_accept_filters
.
empty
())
{
bool
matched
=
false
;
for
(
options_t
::
tcp_accept_filters_t
::
size_type
i
=
0
;
i
!=
options
.
tcp_accept_filters
.
size
();
++
i
)
{
for
(
options_t
::
tcp_accept_filters_t
::
size_type
i
=
0
,
size
=
options
.
tcp_accept_filters
.
size
();
i
!=
size
;
++
i
)
{
if
(
options
.
tcp_accept_filters
[
i
].
match_address
(
reinterpret_cast
<
struct
sockaddr
*>
(
&
ss
),
ss_len
))
{
matched
=
true
;
...
...
src/thread.cpp
View file @
f8f79137
...
...
@@ -355,8 +355,9 @@ void zmq::thread_t::
if
(
!
_thread_affinity_cpus
.
empty
())
{
cpu_set_t
cpuset
;
CPU_ZERO
(
&
cpuset
);
for
(
std
::
set
<
int
>::
const_iterator
it
=
_thread_affinity_cpus
.
begin
();
it
!=
_thread_affinity_cpus
.
end
();
it
++
)
{
for
(
std
::
set
<
int
>::
const_iterator
it
=
_thread_affinity_cpus
.
begin
(),
end
=
_thread_affinity_cpus
.
end
();
it
!=
end
;
it
++
)
{
CPU_SET
((
int
)
(
*
it
),
&
cpuset
);
}
rc
=
...
...
src/ws_encoder.cpp
View file @
f8f79137
...
...
@@ -117,7 +117,8 @@ void zmq::ws_encoder_t::size_ready ()
static_cast
<
unsigned
char
*>
(
_masked_msg
.
data
());
unsigned
char
*
src
=
static_cast
<
unsigned
char
*>
(
in_progress
()
->
data
());
for
(
size_t
i
=
0
;
i
<
in_progress
()
->
size
();
++
i
,
mask_index
++
)
for
(
size_t
i
=
0
,
size
=
in_progress
()
->
size
();
i
<
size
;
++
i
,
mask_index
++
)
dest
[
i
]
=
src
[
i
]
^
_mask
[
mask_index
%
4
];
next_step
(
_masked_msg
.
data
(),
_masked_msg
.
size
(),
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment