1/* Part of SWI-Prolog 2 3 Author: Jeffrey Rosenwald and Jan Wielemaker 4 E-mail: jeffrose@acm.org 5 WWW: http://www.swi-prolog.org 6 Copyright (c) 2012-2013, Jeffrey Rosenwald 7 2018-2020, CWI Amsterdam 8 All rights reserved. 9 10 Redistribution and use in source and binary forms, with or without 11 modification, are permitted provided that the following conditions 12 are met: 13 14 1. Redistributions of source code must retain the above copyright 15 notice, this list of conditions and the following disclaimer. 16 17 2. Redistributions in binary form must reproduce the above copyright 18 notice, this list of conditions and the following disclaimer in 19 the documentation and/or other materials provided with the 20 distribution. 21 22 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 23 "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 24 LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS 25 FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE 26 COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 27 INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 28 BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; 29 LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 30 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 31 LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN 32 ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 33 POSSIBILITY OF SUCH DAMAGE. 34*/ 35 36:- module(udp_broadcast, 37 [ udp_broadcast_initialize/2, % +IPAddress, +Options 38 udp_broadcast_close/1, % +Scope 39 40 udp_peer_add/2, % +Scope, +IP 41 udp_peer_del/2, % +Scope, ?IP 42 udp_peer/2 % +Scope, -IP 43 ]). 44:- autoload(library(apply),[maplist/2,maplist/3]). 45:- autoload(library(backcomp),[thread_at_exit/1]). 46:- autoload(library(broadcast), 47 [broadcast_request/1,broadcast/1,listening/3,listen/3]). 48:- autoload(library(debug),[debug/3]). 49:- autoload(library(error), 50 [must_be/2,syntax_error/1,domain_error/2,existence_error/2]). 51:- autoload(library(option),[option/3]). 52:- autoload(library(socket), 53 [ tcp_close_socket/1, 54 udp_socket/1, 55 tcp_bind/2, 56 tcp_getopt/2, 57 tcp_setopt/2, 58 udp_receive/4, 59 udp_send/4 60 ]). 61 62 63% :- debug(udp(broadcast)). 64 65/** <module> A UDP broadcast proxy 66 67SWI-Prolog's broadcast library provides a means that may be used to 68facilitate publish and subscribe communication regimes between anonymous 69members of a community of interest. The members of the community are 70however, necessarily limited to a single instance of Prolog. The UDP 71broadcast library removes that restriction. With this library loaded, 72any member on your local IP subnetwork that also has this library loaded 73may hear and respond to your broadcasts. 74 75This library support three styles of networking as described below. Each 76of these networks have their own advantages and disadvantages. Please 77study the literature to understand the consequences. 78 79 $ broadcast : 80 Broadcast messages are sent to the LAN subnet. The broadcast 81 implementation uses two UDP ports: a public to address the whole 82 group and a private one to address a specific node. Broadcasting 83 is generally a good choice if the subnet is small and traffic is 84 low. 85 86 $ unicast : 87 Unicast sends copies of packages to known peers. Unicast networks 88 can easily be routed. The unicast version uses a single UDP port 89 per node. Unicast is generally a good choice for a small party, 90 in particular if the peers are in different networks. 91 92 $ multicast : 93 Multicast is like broadcast, but it can be configured to 94 work accross networks and may work more efficiently on VLAN networks. 95 Like the broadcast setup, two UDP ports are used. Multicasting can 96 in general deliver the most efficient LAN and WAN networks, but 97 requires properly configured routing between the peers. 98 99After initialization and, in the case of a _unicast_ network managing 100the set of peers, communication happens through broadcast/1, 101broadcast_request/1 and listen/1,2,3. 102 103A broadcast/1 or broadcast_request/1 of the shape udp(Scope, Term) or 104udp(Scope, Term, TimeOut) is forwarded over the UDP network to all peers 105that joined the same `Scope`. To prevent the potential for feedback 106loops, only the plain `Term` is broadcasted locally. The timeout is 107optional. It specifies the amount to time to wait for replies to arrive 108in response to a broadcast_request/1. The default period is 0.250 109seconds. The timeout is ignored for broadcasts. 110 111An example of three separate processes cooperating in the same _scope_ 112called `peers`: 113 114== 115Process A: 116 117 ?- listen(number(X), between(1, 5, X)). 118 true. 119 120 ?- 121 122Process B: 123 124 ?- listen(number(X), between(7, 9, X)). 125 true. 126 127 ?- 128 129Process C: 130 131 ?- findall(X, broadcast_request(udp(peers, number(X))), Xs). 132 Xs = [1, 2, 3, 4, 5, 7, 8, 9]. 133 134 ?- 135== 136 137It is also possible to carry on a private dialog with a single 138responder. To do this, you supply a compound of the form, Term:PortId, 139to a UDP scoped broadcast/1 or broadcast_request/1, where PortId is the 140ip-address and port-id of the intended listener. If you supply an 141unbound variable, PortId, to broadcast_request, it will be unified with 142the address of the listener that responds to Term. You may send a 143directed broadcast to a specific member by simply providing this address 144in a similarly structured compound to a UDP scoped broadcast/1. The 145message is sent via unicast to that member only by way of the member's 146broadcast listener. It is received by the listener just as any other 147broadcast would be. The listener does not know the difference. 148 149For example, in order to discover who responded with a particular value: 150 151== 152Host B Process 1: 153 154 ?- listen(number(X), between(1, 5, X)). 155 true. 156 157 ?- 158 159Host A Process 1: 160 161 162 ?- listen(number(X), between(7, 9, X)). 163 true. 164 165 ?- 166 167Host A Process 2: 168 169 ?- listen(number(X), between(1, 5, X)). 170 true. 171 172 ?- bagof(X, broadcast_request(udp(peers,number(X):From,1)), Xs). 173 From = ip(192, 168, 1, 103):34855, 174 Xs = [7, 8, 9] ; 175 From = ip(192, 168, 1, 103):56331, 176 Xs = [1, 2, 3, 4, 5] ; 177 From = ip(192, 168, 1, 104):3217, 178 Xs = [1, 2, 3, 4, 5]. 179== 180 181All incomming trafic is handled by a single thread with the alias 182`udp_inbound_proxy`. This thread also performs the internal dispatching 183using broadcast/1 and broadcast_request/1. Future versions may provide 184for handling these requests in separate threads. 185 186 187## Caveats {#udp-broadcase-caveats} 188 189While the implementation is mostly transparent, there are some important 190and subtle differences that must be taken into consideration: 191 192 * UDP broadcast requires an initialization step in order to 193 launch the broadcast listener proxy. See 194 udp_broadcast_initialize/2. 195 196 * Prolog's broadcast_request/1 is nondet. It sends the request, 197 then evaluates the replies synchronously, backtracking as needed 198 until a satisfactory reply is received. The remaining potential 199 replies are not evaluated. With UDP, all peers will send all 200 answers to the query. The receiver may however stop listening. 201 202 * A UDP broadcast/1 is completely asynchronous. 203 204 * A UDP broadcast_request/1 is partially synchronous. A 205 broadcast_request/1 is sent, then the sender balks for a period of 206 time (default: 250 ms) while the replies are collected. Any reply 207 that is received after this period is silently discarded. A 208 optional second argument is provided so that a sender may specify 209 more (or less) time for replies. 210 211 * Replies are presented to the user as a choice point on arrival, 212 until the broadcast request timer finally expires. This 213 allows traffic to propagate through the system faster and provides 214 the requestor with the opportunity to terminate a broadcast request 215 early if desired, by simply cutting choice points. 216 217 * Please beware that broadcast request transactions remain active 218 and resources consumed until broadcast_request finally fails on 219 backtracking, an uncaught exception occurs, or until choice points 220 are cut. Failure to properly manage this will likely result in 221 chronic exhaustion of UDP sockets. 222 223 * If a listener is connected to a generator that always succeeds 224 (e.g. a random number generator), then the broadcast request will 225 never terminate and trouble is bound to ensue. 226 227 * broadcast_request/1 with =|udp_subnet|= scope is _not_ reentrant. 228 If a listener performs a broadcast_request/1 with UDP scope 229 recursively, then disaster looms certain. This caveat does not apply 230 to a UDP scoped broadcast/1, which can safely be performed from a 231 listener context. 232 233 * UDP broadcast's capacity is not infinite. While it can tolerate 234 substantial bursts of activity, it is designed for short bursts of 235 small messages. Unlike TIPC, UDP is unreliable and has no QOS 236 protections. Congestion is likely to cause trouble in the form of 237 non-Byzantine failure. That is, late, lost (e.g. infinitely late), 238 or duplicate datagrams. Caveat emptor. 239 240 * A UDP broadcast_request/1 term that is grounded is considered to 241 be a broadcast only. No replies are collected unless the there is at 242 least one unbound variable to unify. 243 244 * A UDP broadcast/1 always succeeds, even if there are no 245 listeners. 246 247 * A UDP broadcast_request/1 that receives no replies will fail. 248 249 * Replies may be coming from many different places in the network 250 (or none at all). No ordering of replies is implied. 251 252 * Prolog terms are sent to others after first converting them to 253 atoms using term_string/3. Serialization does not deal with cycles, 254 attributes or sharing. The hook udp_term_string_hook/3 may be 255 defined to change the message serialization and support different 256 message formats and/or encryption. 257 258 * The broadcast model is based on anonymity and a presumption of 259 trust--a perfect recipe for compromise. UDP is an Internet protocol. 260 A UDP broadcast listener exposes a public port, which is 261 static and shared by all listeners, and a private port, which is 262 semi-static and unique to the listener instance. Both can be seen 263 from off-cluster nodes and networks. Usage of this module exposes 264 the node and consequently, the cluster to significant security 265 risks. So have a care when designing your application. You must talk 266 only to those who share and contribute to your concerns using a 267 carefully prescribed protocol. 268 269 * UDP broadcast categorically and silently ignores all message 270 traffic originating from or terminating on nodes that are not 271 members of the local subnet. This security measure only keeps honest 272 people honest! 273 274@author Jeffrey Rosenwald (JeffRose@acm.org), Jan Wielemaker 275@license BSD-2 276@see tipc.pl 277*/ 278 279:- multifile 280 udp_term_string_hook/3, % +Scope, ?Term, ?String 281 udp_unicast_join_hook/3, % +Scope, +From, +Data 282 black_list/1. % +Term 283 284:- meta_predicate 285 safely(), 286 safely_det(). 287 288safely(Predicate) :- 289 Err = error(_,_), 290 catch(Predicate, Err, 291 print_message_fail(Err)). 292 293safely_det(Predicate) :- 294 Err = error(_,_), 295 catch(Predicate, Err, 296 print_message_fail(Err)), 297 !. 298safely_det(_). 299 300print_message_fail(Term) :- 301 print_message(error, Term), 302 fail. 303 304udp_broadcast_address(IPAddress, Subnet, BroadcastAddress) :- 305 IPAddress = ip(A1, A2, A3, A4), 306 Subnet = ip(S1, S2, S3, S4), 307 BroadcastAddress = ip(B1, B2, B3, B4), 308 309 B1 is A1 \/ (S1 xor 255), 310 B2 is A2 \/ (S2 xor 255), 311 B3 is A3 \/ (S3 xor 255), 312 B4 is A4 \/ (S4 xor 255). 313 314%! udp_broadcast_service(?Scope, ?Address) is nondet. 315% 316% provides the UDP broadcast address for a given Scope. At present, 317% only one scope is supported, =|udp_subnet|=. 318 319%! udp_scope(?ScopeName, ?ScopeDef) 320 321:- dynamic 322 udp_scope/2, 323 udp_scope_peer/2. 324:- volatile 325 udp_scope/2, 326 udp_scope_peer/2. 327% 328% Here's a UDP proxy to Prolog's broadcast library 329% 330% A sender may extend a broadcast to a subnet of a UDP network by 331% specifying a =|udp_subnet|= scoping qualifier in his/her broadcast. 332% The qualifier has the effect of selecting the appropriate multi-cast 333% address for the transmission. Thus, the sender of the message has 334% control over the scope of his/her traffic on a per-message basis. 335% 336% All in-scope listeners receive the broadcast and simply rebroadcast 337% the message locally. All broadcast replies, if any, are sent directly 338% to the sender via the port-id that was received with the broadcast. 339% 340% Each listener exposes two UDP ports, a shared public port that is 341% bound to a well-known port number and a private port that uniquely 342% indentifies the listener. Broadcasts are received on the public port 343% and replies are sent on the private port. Directed broadcasts 344% (unicasts) are received on the private port and replies are sent on 345% the private port. 346 347% Thread 1 listens for directed traffic on the private port. 348% 349 350:- dynamic 351 udp_private_socket/3, % Port, Socket, FileNo 352 udp_public_socket/4, % Scope, Port, Socket, FileNo 353 udp_closed/1. % Scope 354 355udp_inbound_proxy(Master) :- 356 thread_at_exit(inbound_proxy_died), 357 make_private_socket, 358 thread_send_message(Master, udp_inbound_ready), 359 udp_inbound_proxy_loop. 360 361udp_inbound_proxy_loop :- 362 forall(udp_scope(Scope, ScopeData), 363 make_public_socket(ScopeData, Scope)), 364 retractall(udp_closed(_)), 365 findall(FileNo, udp_socket_file_no(FileNo), FileNos), 366 catch(dispatch_inbound(FileNos), 367 E, dispatch_exception(E)), 368 udp_inbound_proxy_loop. 369 370dispatch_exception(E) :- 371 E = error(_,_), 372 !, 373 print_message(warning, E). 374dispatch_exception(_). 375 376 377%! make_private_socket is det. 378% 379% Create our private socket. This socket is used for messages that are 380% directed to me. Note that we only need this for broadcast networks. 381% If we use a unicast network we use our public port to contact this 382% specific server. 383 384make_private_socket :- 385 udp_private_socket(_Port, S, _F), 386 !, 387 ( ( udp_scope(Scope, broadcast(_,_,_)) 388 ; udp_scope(Scope, multicast(_,_)) 389 ), 390 \+ udp_closed(Scope) 391 -> true 392 ; tcp_close_socket(S), 393 retractall(udp_private_socket(_,_,_)) 394 ). 395make_private_socket :- 396 udp_scope(_, broadcast(_,_,_)), 397 !, 398 udp_socket(S), 399 tcp_bind(S, Port), 400 tcp_getopt(S, file_no(F)), 401 tcp_setopt(S, broadcast), 402 assertz(udp_private_socket(Port, S, F)). 403make_private_socket :- 404 udp_scope(_, multicast(_,_)), 405 !, 406 udp_socket(S), 407 tcp_bind(S, Port), 408 tcp_getopt(S, file_no(F)), 409 assertz(udp_private_socket(Port, S, F)). 410make_private_socket. 411 412%! make_public_socket(+ScopeData, +Scope) 413% 414% Create the public port Scope. 415 416make_public_socket(_, Scope) :- 417 udp_public_socket(Scope, _Port, S, _), 418 !, 419 ( udp_closed(Scope) 420 -> tcp_close_socket(S), 421 retractall(udp_public_socket(Scope, _, _, _)) 422 ; true 423 ). 424make_public_socket(broadcast(_SubNet, _Broadcast, Port), Scope) :- 425 udp_socket(S), 426 tcp_setopt(S, reuseaddr), 427 tcp_bind(S, Port), 428 tcp_getopt(S, file_no(F)), 429 assertz(udp_public_socket(Scope, Port, S, F)). 430make_public_socket(multicast(Group, Port), Scope) :- 431 udp_socket(S), 432 tcp_setopt(S, reuseaddr), 433 tcp_bind(S, Port), 434 tcp_setopt(S, ip_add_membership(Group)), 435 tcp_getopt(S, file_no(F)), 436 assertz(udp_public_socket(Scope, Port, S, F)). 437make_public_socket(unicast(Port), Scope) :- 438 udp_socket(S), 439 tcp_bind(S, Port), 440 tcp_getopt(S, file_no(F)), 441 assertz(udp_public_socket(Scope, Port, S, F)). 442 443udp_socket_file_no(FileNo) :- 444 udp_private_socket(_,_,FileNo). 445udp_socket_file_no(FileNo) :- 446 udp_public_socket(_,_,_,FileNo). 447 448%! dispatch_inbound(+FileNos) 449% 450% Dispatch inbound traffic. This loop uses wait_for_input/3 to wait 451% for one or more UDP sockets and dispatches the requests using the 452% internal broadcast service. For an incomming broadcast _request_ we 453% send the reply only to the requester and therefore we must use a 454% socket that is not in broadcast mode. 455 456dispatch_inbound(FileNos) :- 457 debug(udp(broadcast), 'Waiting for ~p', [FileNos]), 458 wait_for_input(FileNos, Ready, infinite), 459 debug(udp(broadcast), 'Ready: ~p', [Ready]), 460 maplist(dispatch_ready, Ready), 461 dispatch_inbound(FileNos). 462 463dispatch_ready(FileNo) :- 464 udp_private_socket(_Port, Private, FileNo), 465 !, 466 udp_receive(Private, Data, From, [max_message_size(65535)]), 467 debug(udp(broadcast), 'Inbound on private port', []), 468 ( in_scope(Scope, From), 469 udp_term_string(Scope, Term, Data) % only accept valid data 470 -> ld_dispatch(Private, Term, From, Scope) 471 ; true 472 ). 473dispatch_ready(FileNo) :- 474 udp_public_socket(Scope, _PublicPort, Public, FileNo), 475 !, 476 udp_receive(Public, Data, From, [max_message_size(65535)]), 477 debug(udp(broadcast), 'Inbound on public port from ~p for scope ~p', 478 [From, Scope]), 479 ( in_scope(Scope, From), 480 udp_term_string(Scope, Term, Data) % only accept valid data 481 -> ( udp_scope(Scope, unicast(_)) 482 -> ld_dispatch(Public, Term, From, Scope) 483 ; udp_private_socket(_PrivatePort, Private, _FileNo), 484 ld_dispatch(Private, Term, From, Scope) 485 ) 486 ; udp_scope(Scope, unicast(_)), 487 udp_term_string(Scope, Term, Data), 488 unicast_out_of_scope_request(Scope, From, Term) 489 -> true 490 ; true 491 ). 492 493in_scope(Scope, Address) :- 494 udp_scope(Scope, ScopeData), 495 in_scope(ScopeData, Scope, Address), 496 !. 497in_scope(Scope, From) :- 498 debug(udp(broadcast), 'Out-of-scope ~p datagram from ~p', 499 [Scope, From]), 500 fail. 501 502in_scope(broadcast(Subnet, Broadcast, _PublicPort), _Scope, IP:_FromPort) :- 503 udp_broadcast_address(IP, Subnet, Broadcast). 504in_scope(multicast(_Group, _Port), _Scope, _From). 505in_scope(unicast(_PublicPort), Scope, IP:_) :- 506 udp_peer(Scope, IP:_). 507 508 509%! ld_dispatch(+PrivateSocket, +Term, +From, +Scope) 510% 511% Locally dispatch Term received from From. If it concerns a broadcast 512% request, send the replies to PrivateSocket to From. The multifile 513% hook black_list/1 can be used to ignore certain messages. 514 515ld_dispatch(_S, Term, From, _Scope) :- 516 debug(udp(broadcast), 'ld_dispatch(~p) from ~p', [Term, From]), 517 fail. 518ld_dispatch(_S, Term, _From, _Scope) :- 519 blacklisted(Term), !. 520ld_dispatch(S, request(Key, Term), From, Scope) :- 521 !, 522 forall(safely(broadcast_request(Term)), 523 safely((udp_term_string(Scope, reply(Key,Term), Message), 524 udp_send(S, Message, From, [])))). 525ld_dispatch(_S, send(Term), _From, _Scope) :- 526 !, 527 safely_det(broadcast(Term)). 528ld_dispatch(_S, reply(Key, Term), From, _Scope) :- 529 ( reply_queue(Key, Queue) 530 -> safely(thread_send_message(Queue, Term:From)) 531 ; true 532 ). 533 534blacklisted(send(Term)) :- black_list(Term). 535blacklisted(request(_,Term)) :- black_list(Term). 536blacklisted(reply(_,Term)) :- black_list(Term). 537 538 539%! reload_udp_proxy 540% 541% Update the UDP relaying proxy service. The proxy consists of three 542% forwarding mechanisms: 543% 544% - Listen on our _scope_. If any messages are received, hand them 545% to udp_broadcast/3 to be broadcasted to _scope_ or sent to a 546% specific recipient. 547% - Listen on the _scope_ public port. Incomming messages are 548% relayed to the internal broadcast mechanism and replies are sent 549% to from our private socket. 550% - Listen on our private port and reply using the same port. 551 552reload_udp_proxy :- 553 reload_outbound_proxy, 554 reload_inbound_proxy. 555 556reload_outbound_proxy :- 557 listening(udp_broadcast, udp(_,_), _), 558 !. 559reload_outbound_proxy :- 560 listen(udp_broadcast, udp(Scope,Message), 561 udp_broadcast(Message, Scope, 0.25)), 562 listen(udp_broadcast, udp(Scope,Message,Timeout), 563 udp_broadcast(Message, Scope, Timeout)), 564 listen(udp_broadcast, udp_subnet(Message), % backward compatibility 565 udp_broadcast(Message, subnet, 0.25)), 566 listen(udp_broadcast, udp_subnet(Message,Timeout), 567 udp_broadcast(Message, subnet, Timeout)). 568 569reload_inbound_proxy :- 570 catch(thread_signal(udp_inbound_proxy, throw(udp_reload)), 571 error(existence_error(thread, _),_), 572 fail), 573 !. 574reload_inbound_proxy :- 575 thread_self(Me), 576 thread_create(udp_inbound_proxy(Me), _, 577 [ alias(udp_inbound_proxy), 578 detached(true) 579 ]), 580 thread_get_message(Me, udp_inbound_ready, [timeout(10)]). 581 582inbound_proxy_died :- 583 thread_self(Self), 584 thread_property(Self, status(Status)), 585 ( catch(recreate_proxy(Status), _, fail) 586 -> print_message(informational, 587 httpd_restarted_worker(Self)) 588 ; done_status_message_level(Status, Level), 589 print_message(Level, 590 httpd_stopped_worker(Self, Status)) 591 ). 592 593recreate_proxy(exception(Error)) :- 594 recreate_on_error(Error), 595 reload_inbound_proxy. 596 597recreate_on_error('$aborted'). 598recreate_on_error(time_limit_exceeded). 599 600done_status_message_level(true, silent) :- !. 601done_status_message_level(exception('$aborted'), silent) :- !. 602done_status_message_level(_, informational). 603 604 605%! udp_broadcast_close(+Scope) 606% 607% Close a UDP broadcast scope. 608 609udp_broadcast_close(Scope) :- 610 udp_scope(Scope, _ScopeData), 611 !, 612 assert(udp_closed(Scope)), 613 reload_udp_proxy. 614udp_broadcast_close(_). 615 616 617%! udp_broadcast(+What, +Scope, +TimeOut) 618% 619% Send a broadcast request to my UDP peers in Scope. What is either of 620% the shape `Term:Address` to send Term to a specific address or query 621% the address from which term is answered or it is a plain `Term`. 622% 623% If `Term` is nonground, it is considered is a _request_ (see 624% broadcast_request/1) and the predicate succeeds for each answer 625% received within TimeOut seconds. If Term is ground it is considered 626% an asynchronous broadcast and udp_broadcast/3 is deterministic. 627 628udp_broadcast(Term:To, Scope, _Timeout) :- 629 ground(Term), ground(To), % broadcast to single listener 630 !, 631 udp_basic_broadcast(send(Term), Scope, single(To)). 632udp_broadcast(Term, Scope, _Timeout) :- 633 ground(Term), % broadcast to all listeners 634 !, 635 udp_basic_broadcast(send(Term), Scope, broadcast). 636udp_broadcast(Term:To, Scope, Timeout) :- 637 ground(To), % request to single listener 638 !, 639 setup_call_cleanup( 640 request_queue(Id, Queue), 641 ( udp_basic_broadcast(request(Id, Term), Scope, single(To)), 642 udp_br_collect_replies(Queue, Timeout, Term:To) 643 ), 644 destroy_request_queue(Queue)). 645udp_broadcast(Term:From, Scope, Timeout) :- 646 !, % request to all listeners, collect sender 647 setup_call_cleanup( 648 request_queue(Id, Queue), 649 ( udp_basic_broadcast(request(Id, Term), Scope, broadcast), 650 udp_br_collect_replies(Queue, Timeout, Term:From) 651 ), 652 destroy_request_queue(Queue)). 653udp_broadcast(Term, Scope, Timeout) :- % request to all listeners 654 udp_broadcast(Term:_, Scope, Timeout). 655 656:- dynamic 657 reply_queue/2. 658 659request_queue(Id, Queue) :- 660 Id is random(1<<63), 661 message_queue_create(Queue), 662 asserta(reply_queue(Id, Queue)). 663 664destroy_request_queue(Queue) :- % leave queue to GC 665 retractall(reply_queue(_, Queue)). 666 667 668%! udp_basic_broadcast(+Term, +Dest) is multi. 669% 670% Create a UDP private socket and use it to send Term to Address. If 671% Address is our broadcast address, set the socket in broadcast mode. 672% 673% This predicate succeeds with a choice point. Committing the choice 674% point closes S. 675% 676% @arg Dest is one of single(Target) or `broadcast`. 677 678udp_basic_broadcast(Term, Scope, Dest) :- 679 debug(udp(broadcast), 'UDP proxy outbound ~p to ~p', [Term, Dest]), 680 udp_term_string(Scope, Term, String), 681 udp_send_message(Dest, String, Scope). 682 683udp_send_message(single(Address), String, Scope) :- 684 ( udp_scope(Scope, unicast(_)) 685 -> udp_public_socket(Scope, _Port, S, _) 686 ; udp_private_socket(_Port, S, _F) 687 ), 688 safely(udp_send(S, String, Address, [])). 689udp_send_message(broadcast, String, Scope) :- 690 ( udp_scope(Scope, unicast(_)) 691 -> udp_public_socket(Scope, _Port, S, _), 692 forall(udp_peer(Scope, Address), 693 ( debug(udp(broadcast), 'Unicast to ~p', [Address]), 694 safely(udp_send(S, String, Address, [])))) 695 ; udp_scope(Scope, broadcast(_SubNet, Broadcast, Port)) 696 -> udp_private_socket(_PrivatePort, S, _F), 697 udp_send(S, String, Broadcast:Port, []) 698 ; udp_scope(Scope, multicast(Group, Port)) 699 -> udp_private_socket(_PrivatePort, S, _F), 700 udp_send(S, String, Group:Port, []) 701 ). 702 703% ! udp_br_collect_replies(+Queue, +TimeOut, -TermAndFrom) is nondet. 704% 705% Collect replies on Socket for TimeOut seconds. Succeed for each 706% received message. 707 708udp_br_collect_replies(Queue, Timeout, Reply) :- 709 get_time(Start), 710 Deadline is Start+Timeout, 711 repeat, 712 ( thread_get_message(Queue, Reply, 713 [ deadline(Deadline) 714 ]) 715 -> true 716 ; !, 717 fail 718 ). 719 720%! udp_broadcast_initialize(+IPAddress, +Options) is semidet. 721% 722% Initialized UDP broadcast bridge. IPAddress is the IP address on the 723% network we want to broadcast on. IP addresses are terms ip(A,B,C,D) 724% or an atom or string of the format =|A.B.C.D|=. Options processed: 725% 726% - scope(+ScopeName) 727% Name of the scope. Default is `subnet`. 728% - subnet_mask(+SubNet) 729% Subnet to broadcast on. This uses the same syntax as IPAddress. 730% Default classifies the network as class A, B or C depending on 731% the the first octet and applies the default mask. 732% - port(+Port) 733% Public port to use. Default is 20005. 734% - method(+Method) 735% Method to send a message to multiple peers. One of 736% - broadcast 737% Use UDP broadcast messages to the LAN. This is the 738% default 739% - multicast 740% Use UDP multicast messages. This can be used on WAN networks, 741% provided the intermediate routers understand multicast. 742% - unicast 743% Send the messages individually to all registered peers. 744% 745% For compatibility reasons Options may be the subnet mask. 746 747udp_broadcast_initialize(IP, Options) :- 748 with_mutex(udp_broadcast, 749 udp_broadcast_initialize_sync(IP, Options)). 750 751udp_broadcast_initialize_sync(IP, Options) :- 752 nonvar(Options), 753 Options = ip(_,_,_,_), 754 !, 755 udp_broadcast_initialize(IP, [subnet_mask(Options)]). 756udp_broadcast_initialize_sync(IP, Options) :- 757 to_ip4(IP, IPAddress), 758 option(method(Method), Options, broadcast), 759 must_be(oneof([broadcast, multicast, unicast]), Method), 760 udp_broadcast_initialize_sync(Method, IPAddress, Options), 761 reload_udp_proxy. 762 763udp_broadcast_initialize_sync(broadcast, IPAddress, Options) :- 764 option(subnet_mask(Subnet), Options, _), 765 mk_subnet(Subnet, IPAddress, Subnet4), 766 option(port(Port), Options, 20005), 767 option(scope(Scope), Options, subnet), 768 769 udp_broadcast_address(IPAddress, Subnet4, Broadcast), 770 udp_broadcast_close(Scope), 771 assertz(udp_scope(Scope, broadcast(Subnet4, Broadcast, Port))). 772udp_broadcast_initialize_sync(unicast, _IPAddress, Options) :- 773 option(port(Port), Options, 20005), 774 option(scope(Scope), Options, subnet), 775 udp_broadcast_close(Scope), 776 assertz(udp_scope(Scope, unicast(Port))). 777udp_broadcast_initialize_sync(multicast, IPAddress, Options) :- 778 option(port(Port), Options, 20005), 779 option(scope(Scope), Options, subnet), 780 udp_broadcast_close(Scope), 781 multicast_address(IPAddress), 782 assertz(udp_scope(Scope, multicast(IPAddress, Port))). 783 784to_ip4(Atomic, ip(A,B,C,D)) :- 785 atomic(Atomic), 786 !, 787 ( split_string(Atomic, ".", "", Strings), 788 maplist(number_string, [A,B,C,D], Strings) 789 -> true 790 ; syntax_error(illegal_ip_address) 791 ). 792to_ip4(IP, IP). 793 794mk_subnet(Var, IP, Subnet) :- 795 var(Var), 796 !, 797 ( default_subnet(IP, Subnet) 798 -> true 799 ; domain_error(ip_with_subnet, IP) 800 ). 801mk_subnet(Subnet, _, Subnet4) :- 802 to_ip4(Subnet, Subnet4). 803 804%! default_subnet(+IP, -NetWork) 805% 806% Determine the default network address from an IP address. This 807% classifies the network as class A, B or C. 808% 809% @see https://docs.oracle.com/cd/E19504-01/802-5753/planning3-78185/index.html 810 811default_subnet(ip(A,_,_,_), ip(A,0,0,0)) :- 812 between(0,127, A), !. 813default_subnet(ip(A,B,_,_), ip(A,B,0,0)) :- 814 between(128,191, A), !. 815default_subnet(ip(A,B,C,_), ip(A,B,C,0)) :- 816 between(192,223, A), !. 817 818multicast_address(ip(A,_,_,_)) :- 819 between(224, 239, A), 820 !. 821multicast_address(IP) :- 822 domain_error(multicast_network, IP). 823 824 825 /******************************* 826 * UNICAST PEERS * 827 *******************************/ 828 829%! udp_peer_add(+Scope, +Address) is det. 830%! udp_peer_del(+Scope, ?Address) is det. 831%! udp_peer(?Scope, ?Address) is nondet. 832% 833% Manage and query the set of known peers for a unicast network. 834% Address is either a term IP:Port or a plain IP address. In the 835% latter case the default port registered with the scope is used. 836% 837% @arg Address has canonical form ip(A,B,C,D):Port. 838 839udp_peer_add(Scope, Address) :- 840 must_be(ground, Address), 841 peer_address(Address, Scope, Canonical), 842 ( udp_scope_peer(Scope, Canonical) 843 -> true 844 ; assertz(udp_scope_peer(Scope, Canonical)) 845 ). 846 847udp_peer_del(Scope, Address) :- 848 peer_address(Address, Scope, Canonical), 849 retractall(udp_scope_peer(Scope, Canonical)). 850 851udp_peer(Scope, IPAddress) :- 852 udp_scope_peer(Scope, IPAddress). 853 854peer_address(IP:Port, _Scope, IPAddress:Port) :- 855 !, 856 to_ip4(IP, IPAddress). 857peer_address(IP, Scope, IPAddress:Port) :- 858 ( udp_scope(Scope, unicast(Port)) 859 -> true 860 ; existence_error(udp_scope, Scope) 861 ), 862 to_ip4(IP, IPAddress). 863 864 865 866 /******************************* 867 * HOOKS * 868 *******************************/ 869 870%! udp_term_string_hook(+Scope, +Term, -String) is det. 871%! udp_term_string_hook(+Scope, -Term, +String) is semidet. 872% 873% Hook for serializing the message Term. The default writes 874% =|%prolog\n|=, followed by the Prolog term in quoted notation while 875% ignoring operators. This hook may use alternative serialization such 876% as fast_term_serialized/2, use library(ssl) to realise encrypted 877% messages, etc. 878% 879% @arg Scope is the scope for which the message is broadcasted. This 880% can be used to use different serialization for different scopes. 881% @arg Term encapsulates the term broadcasted by the application as 882% follows: 883% 884% - send(ApplTerm) 885% Is sent by broadcast(udp(Scope, ApplTerm)) 886% - request(Id,ApplTerm) 887% Is sent by broadcast_request/1, where Id is a unique large 888% (64 bit) integer. 889% - reply(Id,ApplTerm) 890% Is sent to reply on a broadcast_request/1 request that has 891% been received. Arguments are the same as above. 892% 893% @throws The hook may throw udp(invalid_message) to stop processing 894% the message. 895 896%! udp_term_string(+Scope, +Term, -String) is det. 897%! udp_term_string(+Scope, -Term, +String) is semidet. 898% 899% Serialize an arbitrary Prolog term as a string. The string is 900% prefixed by a magic key to ensure we only accept messages that are 901% meant for us. 902% 903% In mode (+,-), Term is written with the options ignore_ops(true) and 904% quoted(true). 905% 906% This predicate first calls udp_term_string_hook/3. 907 908udp_term_string(Scope, Term, String) :- 909 catch(udp_term_string_hook(Scope, Term, String), udp(Error), true), 910 !, 911 ( var(Error) 912 -> true 913 ; Error == invalid_message 914 -> fail 915 ; throw(udp(Error)) 916 ). 917udp_term_string(_Scope, Term, String) :- 918 ( var(String) 919 -> format(string(String), '%-prolog-\n~W', 920 [ Term, 921 [ ignore_ops(true), 922 quoted(true) 923 ] 924 ]) 925 ; sub_string(String, 0, _, _, '%-prolog-\n'), 926 term_string(Term, String, 927 [ syntax_errors(quiet) 928 ]) 929 ). 930 931%! unicast_out_of_scope_request(+Scope, +From, +Data) is semidet. 932 933%! udp_unicast_join_hook(+Scope, +From, +Data) is semidet. 934% 935% This multifile hook is called if an UDP package is received on the 936% port of the unicast network identified by Scope. From is the origin 937% IP and port and Data is the message data that is deserialized as 938% defined for the scope (see udp_term_string/3). 939% 940% This hook is intended to initiate a new node joining the network of 941% peers. We could in theory also omit the in-scope test and use a 942% normal broadcast to join. Using a different channal however provides 943% a basic level of security. A possibe implementation is below. The 944% first fragment is a hook added to the server, the second is a 945% predicate added to a client and the last initiates the request in 946% the client. The excanged term (join(X)) can be used to exchange a 947% welcome handshake. 948% 949% 950% ``` 951% :- multifile udp_broadcast:udp_unicast_join_hook/3. 952% udp_broadcast:udp_unicast_join_hook(Scope, From, join(welcome)) :- 953% udp_peer_add(Scope, From), 954% ``` 955% 956% ``` 957% join_request(Scope, Address, Reply) :- 958% udp_peer_add(Scope, Address), 959% broadcast_request(udp(Scope, join(X))). 960% ``` 961% 962% ``` 963% ?- join_request(myscope, "1.2.3.4":10001, Reply). 964% Reply = welcome. 965% ``` 966 967unicast_out_of_scope_request(Scope, From, send(Term)) :- 968 udp_unicast_join_hook(Scope, From, Term). 969unicast_out_of_scope_request(Scope, From, request(Key, Term)) :- 970 udp_unicast_join_hook(Scope, From, Term), 971 udp_public_socket(Scope, _Port, Socket, _FileNo), 972 safely((udp_term_string(Scope, reply(Key,Term), Message), 973 udp_send(Socket, Message, From, [])))