Subject: publish / subscribe

From: [email-masked] (Dan Nessett)
Date: Mon, 28 Jun 1993 12:54:12 -0800

I am sending this to the context mailing list, eventhough I am not on it,
because it needs to be discussed before presentation to the full committee.
I am about to leave for a month and won't be able to respond to questions.
My intention is that it will spark some discussion, rather than be the
ultimate solution. If you copy me on the discussion, I will be able to
comment when I get back at the end of July.

Cheers,

Dan

-----------------

It occured to me Friday after the MPI meeting was over that the publish /
subscribe mechanism was trying to solve a problem similar to that of connecting
two virtual address spaces. In Unix this is achieved through the shared memory
system calls, shmget, shmctl, shmat and shmdt. While much of the detailed
functionality of these system calls is not pertinent, I think the concepts
can help us understand how to "attach" two groups, perhaps without a common
ancestor, into one. In order to discuss the similarities and differences
between connecting groups and connecting virtual address spaces, let me
outline a model of groups that facilitates this.

Groups are a kind of virtual address space. The rank within a group specifies
a message pool to which I wish to send messages and from which I wish to
receive messages. They differ from conventional virtual address spaces in
that a process can possess more than one, in fact it can possess many.
When I wish to send a message to another message pool, I specify a group
id, a rank and a context (the first and last items being encapsulated
into a communicator). The rank is a "virtual address" or, if you wish, a
"virtual name" of the message pool in the scope of the group specified by
the group id. The same rank can mean quite different message pools in the
scope of different group ids. This is the major reason why a group is
conveniently viewed as a virtual address or virtual name space. Collective
communication operations do not specify a rank, since it is implicit that
they wish to interact with all message pools within the group.

I use the concept of message pool here, which I acknowledge is non-standard,
because of the confusion that occurs when speaking of processes and
interprocess communication. Processes may or may not consist of one
thread of control. Thus, to speak of a process sending to or receiving
from another process can become confusing, since in some contexts I wish
to discuss threads sending and receiving. My solution is to view the
target of communication operations as a message pool. Processes/threads
carry out interprocess communication by operating on message pools. They
operate on their own message pool by receiving from it and the operate
on other process/thread message pools by sending to them. In a single-threaded
environment, we generally use the term process and there is a one-to-one
mapping between them and message pools. In a multi-threaded environment,
the threads share a common message pool (as they share a common address
space) and the one-to-one mapping no longer exists.

Groups have several characteristics that set them appart from the traditional
concept of a process virtual address space. One already has been mentioned,
i.e., a process may possess more than one group, while it possesses only
one memory address space (actually, this is not quite true, since a
segmented virtual address space can be considered to be the union of a
number of different address spaces, but that is an aside).  Since processes
possess only one address space, they don't have a need to create another.
Also, when they wish to share memory with another process, they map part
of their existing address space into that of the other process. Our model
seems to be that sharing of separately created group address spaces results
in the creation of a new group. Finally, when the address space of one
process is mapped into that of another process, an address that maps to a
common word in the shared space may be different in one process from
that in another process.

Groups, on the other hand, are created by processes to organize their
communication. Since the pattern of process communication within an
application can be quite complex, this may require the creation of several
groups. However, when two processes reference the same rank within a
group, they want to refer to the same message pool. This allows them
to communicate rank information to each other and use it in a consistent
manner.

The last point has significant implications when processes wish to join two
groups into one. In particular, the combined group must map ranks to the same
message pool for all processes. This causes some problems when the join
operation is executed. I think much of the asymmetry in the existing
publish / subscribe mechanism derives from these problems.

The remainder of this message is a proposal to modify the publish / subscribe
mechanism so that it results in the creation of a group by all involved
processes that consistently maps ranks to message pools. It utilizes the
idea that groups are virtual address spaces and that the creation of a
combined group is very similar to sharing memory space.

Combining two groups is carried out in two steps. One process within each of
the two groups publishes a name for the group. The publication operation
conveys not only a character string name representing a published name for
the group (NB: nothing prevents the publication of the same group under
different names), but also the number of processes (message pools) in the
group. The publish operation returns a unique publication id that is used
in a subsequent subscribe operation. The process that executes the publish
communicates the publication id to the other members in its group.

To complete the join, each process in both groups executes a subscribe
operation. The arguments to the subscribe are the name of a published group and
a publication id. The name represents the "other" group, while the publication
id represents the group to which the process executing the subscribe belongs.
The subscribe is a barrier synch operation. When it completes it returns a
new group id that represents the new joined group. The ranks in the joined
group represent message pools in both the original group to which the
process belonged and message pools in the "other" group.

For example, consider two groups A and B with the following characteristics.
Group A consists of 7 message pools and group B consists of 3 message pools.
To keep the description short, I will assume that we are working in a single
threaded environment and so I will use the terms message pool and process
synonymously. Process 2 in group A executes a publish operation on A using
the name "Big group" and receives the publication id 10. Process 1 in group
B executes a publish operation on B using the name "Small group" and receives
the publication id 90. Subsequently, all processes in group A execute a
subscribe with the parameters "Small group" and publication id 10. Similarly,
all processes in group B execute a subscribe with the parameters "Big group"
and the publication id 90. After all processes execute the subscribe
operation, each is returned an id of a group with 10 members, 7 of which
represent the processes in group A and 3 of which represent the processes
in group B.

Some may be concerned about how the publish / subscribe service could be
implemented and how it would interact with the send / receive operations.
Of course, MPI does not describe implementation, since it is an interface
definition. However, in order that people can feel comfortable with
the interface operations described above, let me outline one possible
implementation as an existence proof that efficiency is possible.

First, note that publish / subscribe can be implemented either with a
central server or as a distributed service. The central server implementation
would accept publish requests, create a unique publication id and table
the appropriate information in association with the publication id.
When a process executes a publication request, the group id is used to compute
a list of message pool identifiers that belong to the group. This list is
sent along with the publication name to the central server, which creates
an entry in its data structures, creates a unique publication id and associates
the list of message pool identifiers, the name and the publication id so
it can search on either the name or publication id subsequently and find
the other information. It then returns the publication id to the requesting
process.

A subscription request requires the central server to look up the number of
ranks in the group identified by the name and by the one identified by the
publication id. It adds those numbers together to determine the number of
message pools in the combined group, returns this number with a list of
message pool identifiers that represent the members in the combined group
and records the message pool of the process executing the subscribe as
having completed the operation. When all processes in the combined group
have executed the subscribe, the central server sends a message to each of
them, releasing them from the barrier synch.

Discussion of a distributed service requires the establishment of some
terminology. Since the publication / subscribe service allows groups
created by processes without a common ancestor to join, there must be
a way for the MPI implementations of these processes to communicate. For
example, two parallel programs running on the same MPP machine may not
have a common ancestor, but the machine generally will allow processes
running on a node to communicate with other processes on other nodes
irregardless of whether they have a common ancestor. The MPI implementation
uses lower level identifiers, such as node id and UNIX pid, to carry
out its associated semantics. Similarly, in a distributed system, an
MPI implementation will use IP addresses and TCP ports as its lower
level identifiers. I will call these lower level identifiers and the
communication primatives that operate on them the universe of communication.

In order to control scaling problems in large systems, the publish /
subscribe caller must limit the universe of communication. For example, if
MPI is being used in a distributed system based on the internet as the
universe of communication, a publish / subscribe activity would have to
contact every process executing within it. Obviously, this is unreasonable.
There could be either a environmental management call or a separate
publish / subscribe call that establishes this limit. For example, the call
could establish the universe of communication to be all processes within
a particular partition on an MPP machine. In a distributed environment,
the call could establish a particular subnet address or a set of subnet
addresses to limit the universe.

I will assume that there is a broadcast facility in the universe
of communication. For an MPP this seems obvious. For a distributed system
this is less obvious, but much work is on-going at the present time to
support a scalable broadcast facility in distributed systems in order to
support various forms of teleconferencing.

When a process executes a publish operation, it picks a unique integer and
concatenates it with its unique address within the universe of communication.
This becomes the publication id. It communicates this id and the published
name to the other members in the group. When a subscribe operation is
executed, the process broadcasts over the universe of communications the
other group's publication name, its own publication id and a list of members
in its group. This list must be expressed using the identifiers in the
universe of communication.

Each process blocks within the subscribe operation until it receives a
subscription message with either the publication name or its publication id.
When the message contains the publication id, it must have come from some
process within its own group. It marks that process off the list of processes
waiting on the barrier synch and waits for another message (unless the
completion criterion described below is met). When the message contains the
publication name, if this is the first such message, it extracts the list
of members in the other group and forms a concatenated group. Some rule must
be used so that all processes choose the same ranking. This could be on the
basis of numerical or lexigraphic ordering of the identifiers in the
universe of communication. The process then marks off the source of the
message from the list of processes waiting for the barrier synch and waits
for another message. If this was not the first message with the publication
name, it simply marks off the process. When all processes are marked off
of the list, the subscribe completes and the call returns.

One problem with this scheme is that all processes must process every subscribe
even if they are not involved. To mitigate this performance hit, a combination
of the central server and distributed server implementation strategies is
possible. Processes can be bound to "local" central servers with which
they execute the central server protocol. The central servers can then
execute the distributed protocol (or some suitably modified version of it)
to complete the publish / subscribe operation. This offloads interfering
publish / subscribe activity onto a server with only that function.