This system has been defined in USL; and automatically analyzed and
automatically generatated/resource allocated (RAT)ed to C code using the
001 Toolsuite.
The client/server
system uses the OMap Editor to ask the user for input (System Inputs) and
responds with the results (System Outputs). In this client/server example,
the user enters two books (book1 and book2).
The server then counts
the total number of occurrences of words in the two documents. To do this
the server calls the function ngram_test which is an FMap in another
library that implements a MapReduce like algorithm (not shown in detail).
The Map Phase of MapReduce
makes a set of key-value pairs following a 1-ngram approach. This means
that each word becomes a key with the value of “1”. The set is really
a bag of word-value pairs in which each word as a key may be duplicated;
but each word as a key will have the value “1”.
The Reduce Phase
will collect each of the duplicated keys into a unique keyed (word) set of
values (or counts). The counts are collected for all the unique keyed set
elements (i.e., words). Then, for each unique keyed word, the set of
counts are summed to get the total number of counts for each keyed word.
This set of unique keyed value pairs, as word-count pairs, is converted to
a set of msgs(OSetof)msg(str) in the format: word:count.
client starts a server on the
same host as the client, waits for two seconds and then requests the
server to make a two-way communication channel for sending and receiving
data. In our example, the variables hearFrom and talkTo are the CPort
(connected port) ends for communication between the client and server:
ssh_exec merges
together the ssh command, the user and host name and the path/executable,
pathexe. This command, cmd, is then executed using the spawnNoWait
function to start the server either locally or on a foreign machine
designated for hosting the server.
connect_to_server makes a host INet address, encoded as
a binary integer, and adds the port "5000" to it to get the server's INet
address and listening port. Then the client issues a requestConnext:CPort
function to initiate the server's listening port to request a communication
channel for talking to the server.
send_msgs sets up a
synchronous SELECT system call to talk to the server with the client's end
of communication channel set up by the server (and accessed by the client
with requestConnect:CPort). If the client is notified (through SELECT)
that it is ok to talk, the client sends its messages, ms0, to the server by
converting those messages to an OMap form and then to a STR string form to
marshall the data to the server (who is synchronously waiting for them).
write_error_select sets up the client for writing to the cport0 CPort
channel provided by the server. To do this, the cport0 descripter is added
to the file descriper sets for error and write (using the add:CPORT,fdset
functions). The SELECT:fdset function will return when either of the write
or error fdsets are signaled via the cport0 descripter. The form for the
SELECT primitive operation is:
is_present_in_fdset checks to see if the CPort, cport,
is in the descripter set, fdset. If it is present, a TRUE value is
returned for the Boolean, b0; otherwise a FALSE or REJECT value is
returned. Here, if b0 is REJECT, we set the value of b to be FALSE.
id_d3 selects a0 to become a and the objects 1, 2 and
3 are deallocated.
Interface: a:Any=id_d3(a0:Any,1:Any,2:Any,3:Any)
()
1.1.3.2 sendOMap.
sendomap is a universal (polymorphic) operation that takes as a
parameter, OSETOFTYPE. Given the particular OSETOFTYPE for set, set will
be converted from its strongly typed OSETOFTYPE viewpoint to its weakly
typed OMap object viewpoint, om. The om OMap set object is further
transformed to a Str string, omstr0. The size of the omstr0 object is
determined; the size converted to a string; and then prefixed to the omstr0
object. At this point, we have a string, omstr, with the format:
size#OMap-as-a-string-object
This string object will be marshalled
(sent) to the other end of the CPort communication channel, cport0, by the
function, Send_a_str_CPort.
receive_msgs sets up a
SELECT system call using the function read_error_select. In our current
case, the wait Nat value is REJECT indicating the our recieve process is to
be synchronous. If the value of the Boolean, r, is "TRUE" for reading,
then the CAN_RECEIVE sub-function is performed. An OMap string is received
from the CPort, cp0, by function Receive_a_Str_CPort (through data
marshalling and used by either the client or server from either end of
their communication channel). Once the OMap data STR string is received,
it is converted to a msgs (OSetOf) object containing the transmitted
information.
server starts on the host
designated by the client (in our example, on the same host as the client)
and creates a listening port/INet address. It then sets up the SELECT
system call for listening--waiting for a client. When a client issues
RequestConnect:CPort on the port/INet address (see connect_to_server called
from client) the SELECT function returns. If all goes well, the server
connects to the client using the awaitconnect:lport function (see
connect_to_client) and attempts to handle the clients request. The server
synchronously receives the clients msgs because FOREVER is set to the
REJECT Nat value. The msgs, ms, are sent to the function to handle the
client's request; here, ngram_test. Once the requested function returns,
the server sends the resulting msgs, sums, back to the client by converting
the sums msgs to an OMap and then marshalling the OMap as a STR string
object to the client. After this is done, the server exits.
setup_listening_inet_port sets up the servers INet address and port for
listening for a client. The server which is running now in a user account
on a host machine, gets its host machine hostname, makes a binary Integer
INet address from it (host), and adds the listening port to the address,
"5000". A "1000" byte buffer for messages arriving on the listening port
is set.
set_select_for_listening sets the listening LPort, listenOn, to wait
(FOREVER) for any client that knows its INet/port address. After setting
up the descripter sets read and error to be signaled when listenOn is
activated, the time_SELECT_on_fdsets function is used to perform listening
(synchronously listening in our case because of the k:REJECT,nat function,
see select:fdset in the operation read_error_select for details).
time_select_on_fdsets gets the descripter setups from
fdset, fdss0, for reading, writing and exception (error), calls the SELECT
system call and puts the response back into fdset, fdss.
viewmore_2 provides access to the two children objects,
1 and 2, contained in the object t.
Interface:
1:Any,2:Any=viewmore_2(t:Any)
()
1.2.5 connect_to_client.
connect_to_client waits (using the primitive function,
awaitconnect:lport) for a client to connect to its INet address listening
port. When there is a client on the listening port, client0 is returned as
the server's end to a communication port for talking and passing data to a
client.
id_d1 selects a0 to become a and the object 1 is
deallocated.
Interface: a:Any=id_d1(a0:Any,1:Any)
()
1.2.6 ngram_test.
ngram_test loads a set of files (simulating the
MapReduce SPLIT function) from a server directory, dir. It makes a set of
name/value pairs (i.e., word:"1") for each word found in the set of
documents. It then executes a REDUCE-PHASE reader function, grouping of
same named keys function (collecting associated value counts into a counts
set) and summing the values for each unique key (i.e., word). Finally,
ngram_test converts the summed set of key/values into a msgs set to present
to the user.