2020-08-03
Central to the implementation of the definitive features of a distributed system are the forms of structuring the data inherent to it. This is due to the fact that the primary constraint on such a system is the distribution of data, along with the essential consideration of message timing between nodes. While the algorithms in this particular statistical system are equally necessary for functionality in the stated aim of modelling and manipulation of large data, they are completely dependent on the structure of the data such algorithms operate upon, hence the primality placed on the system’s information structures.
The first questions asked of a distributed system’s information
structures relate to it’s topology and mechanism of message passing.
This particular system answers that necessarily through it’s context and
aims; that it is currently intended to exist as a largely transparent
platform within R, which necessitates interactivity and other standard R
development practices. Thus, a centralised structure must be chosen for
the system, with the node associated with the user’s R session acting as
the master. This is the form of operation of other distributed packages
currently existing in R, such as sparklyr and
SNOW [1], [2]. This can be contrasted with a
decentralised system as in pbdR and it’s underlying
MPI implementation [3], which requires R programs to be
written in a manner agnostic to the nodes in which they are executed
upon, resulting in the same program being distributed to all nodes in
the system. In this way, the program is not capable of being run
interactively, something undesirable to the goals of our system. The
master will therefore always be local, all other nodes remote.
As it currently stands, the local information structures are entirely described by S3 classes, the instances of which act as references to the payload data being held remotely. These classes are composed as environments, used for their mutable hash table properties, and contain three elements:
locsRserve connections through which the remote
payload data resides in discrete chunks
namesizelocs, describing
the size (as in nrow or length) of the chunk
at each location
This is coupled with a garbage collection system consisting of a hook
to the removal of the reference object through
reg.finalize. Upon triggering the hook, a directive is
issued to all chunks in locs to remove name,
thereby closing the loop between creation and deletion on local and
remote nodes.
The system exists in it’s current form primarily through motivations
of simplicity; minimising complexity in the system until further
additions are required. By themselves, loc and
name are sufficient for referencing any distributed chunks.
size is maintained for the regular need to know lengths of
objects as part of many standard operations, thereby reducing the lookup
cost by keeping the static information locally and directly attached to
the reference.
The mechanism of garbage collection is likewise borne of simplicity and necessity; it requires the least possible steps, and without it, distributed chunks would accumulate remotely with no means of further access if their reference is removed, essentially forming a high-level memory leak.
In spite of, and indeed because of, the simple information structure of the local system, there remain aspects of the design that inhibit the development of important features, many of them essential. In addition, clarification in system semantics has revealed a need for greater focus in areas presently under-served by the system.
A major feature lacking in this system is a global awareness of
existing connections, which can be used in preference to creating new
connections upon instantiation of a distributed object. Take for
example, the act of reading in successive distributed csvs
into the system. The first read takes in file location arguments, among
others, then creates new connections, finally returning a reference. The
next read performs exactly the same actions, and so on. This ignores the
highly likely situation where files are situated in the same locations,
and connections at those locations can be reused, thus potentially
saving from the overhead of extraneous connections and unnecessary data
movement of aligning objects with each other.
Another issue is the closing of connections; as it currently stands, there is no appropriate garbage collection for connections.
The single name for all chunks also cuts out any possibility of having multiple chunks belong to the same object referenced via a singular connection, thereby cutting out a potential mechanism for arbitrary indexing of objects.
Significant enhancements to the system can be attained through
additional structures addressing the present deficits. Principally, the
introduction of a central table of connections will serve as a single
source of truth, avoiding issues of non-knowledge in creation, deletion,
and usage of channels. This would require a change in the structure of
reference objects, and can consist in changing literal
RSclient channel pointers to identifiers to be searched for
in the central location table. In this way it provides a solution in the
manner of the fundamental theorem of software engineering:
All problems in computer science can be solved by adding another level of indirection [4].
The table slots correspoinding to each identifier may also contain relevant information to the connection such as host name, rack, etc., in order to optimise data movement, as well as aid in the decision of whether or not to create new connections for newly read or instantiated objects.
Additional improvements, though unrelated, include changes to the reference classes to allow for globally unique names for each chunk, which will allow the same connection to house multiple chunks of a cohesive distributed object, thereby enabling arbitrary indexing operations. With such changes in structure, garbage collection is able to be enhanced through centralising the objects of garbage collection within the central table of locations.
One potential algorithm for garbage collection could involve marking table elements with chunks to be removed, at their associated channel, as part of a reference garbage collection hook. The marked objects can then be used as part of a directive for remote removal at the next convenience. This can be combined with a reference counter of the number of extant objects at the referent environment of each channel; upon complete emptying of the environment, signified by a counter of zero, that channel itself may then be closed and removed.
Most other distributed systems in R require manual specification of a
cluster that then operates either in the background or as an object that
must be retained and manipulated. What is described here bears closer
resemblance to a file system than any particular distributed R package,
with particular relation to the UNIX file system [5], [6]. In the
UNIX file system, files contain no additional information
beyond what is written to them by the user or file generation program.
Directories are also files and consist solely of a regular textual
mapping from a file name to it’s entry (inode) in a central
system table (ilist). The inode contains metadata associated
with a file such as access times and permissions, as well as the
physcical address of the file on disk.
To analogise, references in our system are equivalent to directories.
They provide a mapping from connection names (files) to their entries
(inodes) in a central table (ilist). Furthermore, the table entry
contains an RSclient pointer, analogous to a disk address,
as well as metadata. The form of the metadata differs due to separate
priorities; a list of chunk names in the place of permission bits, etc.
In theory this also allows copies of references to behave as hard links,
though this will introduce major issues involving synchronisation, and
is therefore be avoided for now. This form of garbage collection bears
some resemblance to the file system garbage collection as well, in that
inodes count the number of links to them, issuing a removal directive at
zero links, though our system supplements this through collecting
specific names of chunks for second degree removal. In this manner, the
“marking” via name collection is closer to the method of marked garbage
collection, in conjunction with reference counting [7].
The remote end of the system is the simplest component of the entire
setup. Currently, each remote R process is hosted through
RServe, and accessed through RSclient on the
local end. The remote R process holds chunks of data in it’s global
environment, and performs whichever operations on that data as are
directed to it from the master R session. The data possesses no more
structure than what was already in the chunk following reading,
operation upon, or reception by the node. This has again been due to
reasons of simplicity, as no presuppositions of structure suggested
themselves at the outset.
The system works very well for something general purpose. However, it ignores much of the structure inherent in common primitive R objects such as vectors. For example, to numerically index elements of a distributed vector, an indexing algorithm currently translates the index numbers into node-specific indices, and forwards those translated indices on as part of a call to the relevant nodes. This sees issue when there are disparate elements at a particular node selected between the elements of other nodes, and the mechanism for numerical translation breaks down.
A potential mechanism for improvement is to attach index attributes corresponding to the overall index to the chunks. In combination with remotely-run routines, the local session simply needs to send out a request for particular indices to all of it’s connections, and they can work out themselves which elements, if any, they correspond to, returning a vector of elements matched, for us in the creation of a new reference locally.
This would certainly solve the problem, however it may be redundant
to simply allowing local index translation to account for multiple
chunks at a single connection (as described in section sec. 3.4). It certainly uses significantly more
messaging bandwidth, though by distributing processing of index
translation across nodes, it may be faster in practice. In addition, the
additional structure forced on data chunks by attaching indices is
somewhat contrary to the lack thereof in the analogous UNIX
filesystem described in section sec. 3.5.
Further work involves the actual implementation and assessment of the
proposed information structures, as part of a general rewrite.
Additional research may also involve other garbage collection systems,
with especial interest in file systems, such as those of the Inferno and
Plan 9, distributed operating systems borrowing heavily from
UNIX [8], [9].