Ship Detection: Parameter Server Variant

In regard to maritime object detection, the more generic term of ”vessel” is unused as it refers to other objects that are not distinctly ships, such as floating docks and canoes. Locating ships in aerial imagery is typically performed using Synthetic Aperture Radar (SAR) data [11] [10]. SAR receives high degrees of accuracy since the data provided are not affected by sunlight or clouds. However, utilizing satellite optical imagery provides more identifiable information about ships [10]. Ship detection in coastal and open bodies of water has a wide range of applications such as economic, security, environmental, and ecological [10]. Previous work implementing object segmentation models (focused on U-Net) have seen accuracies reach  57% [21] and for the same dataset used in this work,  85% [9]. Another approach implements other model types, as well as their own, that are typically used for object detection (YOLO, Faster R-CNN, SSD, etc) ranged from 62.4% to 88.3% accuracies [13].

Distributed systems can be utilized in order to improve the computational power of a system [24]. By horizontally scaling the number of compute nodes, the amount of computational power increases. This allows systems to process a larger amount of data/tasks as compared to a single machine. Vertically scaling is typically more expensive, thus the focus on horizontal scaling. With the increase of nodes and compute power, there is also an increase in the issues surrounding distributed systems.

The general scheduling problem can be written as $\alpha|\beta|\gamma|\mu$ [4]. The first attribute, $\alpha$, is a single entry and description of the execution environment. The attribute $\beta$ can have from zero to multiple entries, and describes what is the processing to be done. The $\gamma$ attribute details the description of the objective to be optimized. Finally, $\mu$ is the charging criteria of models adopted in the system: hourly based charge, an auction system, per-transaction charge, data-access charge, and so on.

By distributing computation, there is a need for oversight and regularization of tasks for the system to process, this is where scheduling comes in. Scheduling is a decision-making process that is used on a regular basis in many manufacturing and services industries. It deals with the allocation of resources to tasks over a given time period and its goal is to optimize one or more objectives [17]. Implementing a distributed system requires cost for hardware support and agreements on service expectations. Optimizing tasks through proper scheduling helps reduce the overall cost of computation while increasing the value customers receive.

A system is centralized when it contains a central node or process that performs the necessary delegations to other nodes — this node is called the primary node [4]. The other delegated nodes are considered workers. Decentralized distributed systems offer a more obscure abstraction of primary/worker relations. A node may be a primary of a centralized subset of the network and simultaneously exist as a worker of the larger network, or it may be a worker in its entirety while existing at the same level as a primary with its own worker pool. The primary nodes communicate between each other in order to achieve the system goal.

Another type of distributed system architecture falls into a hierarchical structure. This hierarchical distributed system offers a tiered layer where a node may communicate with only its predecessor or successor (either one level above or below).

The organization of the architecture may play a role in the scope of the scheduler. Not only should we consider the architecture connectivity of primary, workers, and their resources but also policies and access controls when implementing a scheduler. In a cloud based environment, these access controls limit the network and physical resources a worker can communicate with.

A cluster, in this realm, is a collection of dedicated computational hardware that can communicate; usually over a dedicated private network (but distributed systems also exist over public networks with heightened security protocols) [4]. Careful consideration towards task allocation can have immediate effects on the efficiency of a system. The type of system is what dictates the best scheduling approach. In a synchronous scheduling of tasks, the network is only as fast as its slowest worker. The same applies to asynchronous architectures, but typically with a faster rate of overall system completion. The rate of completion also depends on the nature of the task itself. If tasks are not dependent on each other, then the system may progress concurrently. However, if tasks do depend on each other, then the system may experience a varied progress rate where at times it may be zero.

A task is a logical unit that makes progress toward a unified goal of a system [3]. One such set of tasks could be images that a distributed model is learning or predicting on. Another such set of tasks could be a set of computations offloaded to multiple nodes in small chunks and sent off to workers to be processed. A task may end at the worker node — perhaps a state change — or a response may be communicated back to the primary. There are two types of tasks: dependent and independent.

Dependent tasks rely on the response from a task ahead of itself before they can be executed. These tasks are better suited for a synchronous system that waterfalls from one task to the next depending on dependency. The system makes progress at a more determinable rate due to the predictive analysis of task succession.

Independent tasks can be run in parallel and/or concurrently. These tasks run best in an asynchronous system. The downfall to consider is resource contention if a particular set of tasks require the same network resource (say, accessing a file on a shared storage) — this situation has workarounds: eventual consistent databases/servers. The system makes unified progress at an undetermined rate as tasks may run simultaneously toward the system goal.

The process of managing task allocation, to where and to whom, is the responsibility of a scheduler. The scheduler of a distributed system performs akin to the process scheduler on any operating system.

Task-graph scheduling is considered an NP-hard problem [23]. Scheduling tasks from the primary to workers can occur at multiple stages. There can be global scheduler that directs all tasks to all connected workers. There can also be local schedulers that handle incoming and outgoing tasks/replies on both the primary and worker nodes. Organizing this dynamic scale of events is a major component of the system efficiency. In a centralized distributed system, it is more common to see a global scheduler managing the allocation of resources and tasks between primary and workers. However, in a decentralized distributed system, there may exist multiple schedulers that are responsible for part of the overall system. Management of this decentralized state may be distributed over resources. Each scheduler in a decentralized system can be considered the same as that of the scheduler in a centralized system. Schedulers are considered to follow one of two main categories: List or Cluster scheduling [3]. List scheduling is composed of two parts: a priority/ordering of tasks and a mapping to a processing node. Cluster scheduling extends List scheduling by mapping collections or sets to processing nodes.

Infrastructure as a Service (IaaS), Software as a Service (SaaS), and Platform as a Service (PaaS) are common models for user services and each have their own scheduling requirements [4].

IaaS — the most popular in cloud environments — offers physical and virtual environments to deploy computations and/or applications at scale on a public or private based setting. Some of these features are restricted through access controls that are subject to cost-per-use or subscription-based. For IaaS, the scheduler is responsible for “allocating virtual machine requests on the physical [architecture]” [4]. For PaaS and SaaS, the scheduler is responsible for allocating the aforementioned applications and/or computational chunks to workers in order to be executed/served.

Part of optimizing the objectives of cloud-based scheduling includes the Service Level Agreement (SLA) of users with their end clients (workers). The throughput and latency of a system directly affects users first and third party SLA agreements — these SLA agreements can be synonymous to Quality of Service (QoS) expectations.

To achieve high SLA/QoS, a scheduler needs to be robust in order to reduce execution time and cost. A robust scheduler must be able to handle a change in application topology (new instances dynamically being added and removed from the pool), resource/software configurations, input data, and data sources. This is not a comprehensive list of possible dynamics a scheduler should be robust to. One of the biggest nuances of distributed systems is the ability of a schedulers’ robustness against failure occurrences.

Handling failure occurrences is a major proponent of distributed systems. The scheduler must be able to salvage, redistribute, and/or re-implement these occurrences in order to maintain system QoS. Granted, if a failure occurrence is large enough, there will be interruptions in service. However, if a failure occurrence is small scale and handled properly, an end user may never even notice.

There are many kinds of scheduling algorithms that each satisfy different constraints of a system. Choosing the most effective algorithm ensures the optimization of the system goals are achieved.

During prediction, production models will most likely be served in an app on a scalable cluster. This cluster can be either on a heterogeneous or homogeneous system and tasks will be allocated with the above type algorithms. However, Machine/Deep Learning task allocation during training are more specifically scheduled depending on the architecture of the framework. Framework architectures may be decentralized, such as AllReduce, or centralized, such as the Parameter Server. The following two algorithms are focused on the Parameter Server architecture. They aim to reduce the limitations imposed by straggling workers (workers that have fallen behind the others progress rate). Such algorithms include Cyclic or Staircase Scheduling [2].

When provided with tasks, workers sometimes either become corrupt (maybe by a broken process) or another process has started (out of the control of the user and worker node) that demands more computational power that causes the computation of the task to take longer; these workers are referred to as stragglers [16, 2]. Perhaps it is the task itself that is computationally demanding that causes a worker to take longer. In other (probably most) cases, it is the network that bottlenecks task completion communication between the worker and primary [24].

When one (or more) workers begin to experience these disruptions, they fall behind the other workers in terms of progress over assigned tasks — these are considered straggling workers. The product these straggling workers output may become stale if the application requires operations on current data and therefore useless. If stale data is used there is the possibility that it will hinder the progress made by other workers.

It’s fairly intuitive to reason about workers that are persistently straggling — they fell behind and are in a constant state of catch up and are persistently falling behind due to lack of computational power/network bottlenecking compared to other workers. Non-persistent types of straggling workers are harder to predict and analyze solutions for [16, 2]. Non-persistent straggling workers are intermittently straggling but may not always be in as much of a deficit as compared to other workers as they may complete a significant portion of the assigned tasks by the time other workers have completed theirs.

The architecture of a parameter-server system, typically centralized, can involve multiple layers of server groups, schedulers, and workers [12]. This architecture is used to leverage more compute power against massive datasets, such as datasets found in industry that can go from TBs to PTs [12]. Attempts to provide better solutions for convergence with parameter servers involve a process that makes use of accumulated gradients [8]. Parameter servers do not typically converge at the same rate as synchronous SGD approaches [7] [25].

Delivering data to the training cycle was handled by a Redis connection client. A module of Redis, RedisAI [19], that within it’s datatype structure can manage tensors via the Redis standard key-value approach is used as the server structure. This is useful to communicate large amounts of tensors back and forth asynchronously since Redis can handle massive amounts of requests - it can easily handle 10,000 clients as of Redis version 2.6 [18]. Leveraging RedisAIs in-memory caching system allows for training data to be delivered rapidly. One downfall to this approach is that task delivery is limited by the network bandwidth as with all network bound communications. However, this delivery service allowed for the training data to be available and in a ready format in memory without the use of file I/O operations. was particularly useful as the single node architecture was trained concurrently with the following parameter server variant with both using the same data server. Also, this method of data delivery allowed for concurrent reads of key-values. This means that, in a parameter server architecture, any worker may read the global model weights concurrently as another worker. This eliminates downtime due to resource locks. In an Azure virtual private network, specific inbound port rules have to be assured to prevent malicious interaction/use from outside users.

Both frameworks utilized a RediaAI server to load chips. One of the most effective ways of serving traditional files for deep learning is using a HDF5 format and served as a comparison to RedisAI. For this experiment, each method of delivery iterated through individual files (both identical Numpy array format) and converted into float tensors that a model could receive. However, the model step was skipped as to just compare file serving methods. The results are summarized in Table 2.

It is intuitive that utilizing the data server is the more effective approach since it processed all 63 thousand chips within a few minutes, whereas HDF5 I/O took over three hours. The use of a data server increases the usage of the GPU to 95-99% consistently as per the nvidia-smi command, whereas file I/O would see dips to 0% (although microseconds) when workers are not able to process an item or are limited by the number of cores versus the batch size in relation to the overhead of file I/O.

Table 21 states the completion time in seconds and cost for all run experiments. The SN architecture made use of a single NC24 virtual machine for training. The PSv made use of a NC24 and four NC6 virtual machines. However, the data server was hosted on the primary node of the PSv architecture - this was a cheaper solution than having it run on a individual virtual machine. Since the SN architecture used the data server, it can also be suggested that the virtual machine that it trained on could also host the data server in an isolated training cycle and would absorb the cost. The cost is calculated based on hourly usage of each virtual machine [14]. In these experiments, the promotional version of the virtual machines were used for a reduced cost while they were available, but the cost will also be calculated for full-price usage. The NC6 Promo cost $0.396/hr and NC24 Promo $1.584/hr. The non-Promo NC6 cost $0.90/hr and non-Promo NC24 cost $3.60/hr.

Since I/O operations are vastly more expensive than in-memory retrieval of data, this is why a data server was chosen over standard image reads for serving up datapoints. Iterating and transforming over the entire 63223 image chips and converting them into learning-ready tensors only took 2 minutes and 32 seconds while the file I/O operations took over 3 hours. This comparison to read and prepare is the reason serving data to models from a data server was the chosen approach. Also, utilizing a data server allowed for uninterrupted GPU usage that allows for a more cost effective approach to training.

It is expected that each of the SN approaches achieve similar times for their functionalities. Consideration toward the batch training time needs to be taken in regard to the maximum and the variance. The small variance indicates that the average batch training time is consistently around the mean time. However, the maximum time is excessively larger than the mean. This is due to the preparation of the dataset at the beginning of each epoch as the architecture implements the built-in (PyTorch) data loader objects, which results in a larger up front time overhead. Intuitively, training on a 20k dataset takes around twice as long as when training with a 10k dataset. The timing records for the 10k dataset over 100 epochs are lower than the other two most likely because the total number of recordings are numerously greater than the 30 epoch training runs. Though, it is interesting that the variance of the time it took the 20k dataset over 30 epoch training run to save the model to the Azure Fileshare is quite large compared to the other two approaches. The high variance of model saving could suggest that there was more fluctuation in network traffic between the compute node and the storage location or internal process contention at the time of this training run and at the end of each epoch when the model is saved since the standard deviation is about 11 seconds for Worker_002 and about 7-9 seconds for the other workers. Though, it is more likely there was internal process contention delaying the initialization of the saving process within the VM that may also be paired alongside network bandwidth that causes such a large discrepancy. That same hypothesis could be applied to the other two approaches also at a much lesser degree since the variance is not very small.

Reading of the global model to initialize the local models is at the beginning of the PSv pipeline. The variance of records is small enough to indicate that there were no major networking or local issues at each read. Reading of the global model is one of the larger portions of overhead in the system. However, operating with a Redis based data server allows for concurrent reads and therefore reducing contention around resources. This larger time is indicative to the models large size for this application.

When the scheduler receives a task request, it delivers the task to a worker which processes it into a batch before performing a forward pass through the model. While the variance across all workers is really small indicating that the mean value is close to the true time, it is notable that worker_003 maximum value was recorded as three times as long as the other workers. This is most likely a rare event (considering the variance is so small), and probably due to some internal process contention. The process of waiting for the task to be delivered and the time taken to process a task request are interoperable and dependent. A reason the worker wait times are bigger than the scheduler process times are due to the scheduler processing worker tasks synchronously as they are received and some workers are forced to wait while other workers are served. Once the worker has received the task, it is read from the data server, built into a batch, and a forward pass is processed. The times taken for processing a task, all with relatively small variances, are all faster than that of the SN architecture. Implementing a PyTorch data loader is convenient, however, it introduces an overhead. This step can be multiplied by the number of accumulations to represent a more accurate ”batch” training time, but it is assessed individually with respect to the SN architecture performance comparison. Implementing a request-deliver type scheduling algorithm reduces contention over resources and the chances of stragglers occurring. Although it takes half as much time to prepare and process each batch, the accompanying overhead from requesting tasks and pushing/pulling global model updates causes the PSv architecture to operate slower than the SN architecture.

Uploading gradients for the large U-Net model varied over the training run as per the larger variance value in Table 15. This deviation from the mean could be attributed to the complexity or size of the update or it could be network bound from uploading a large amount of data. Most likely, it is a combination of both of these things that cause the overhead. The minimum time for each worker suggests an ideal upload traffic load (close to zero) between the worker node and the primary node. Whereas the maximum time is the opposite and worst case scenario. Once pushed, the worker will begin waiting for the primary process to retrieve and update the global model weights before pushing them for the worker to read. The process is synchronous at the primary process as gradients are applied as they come in and recognized at each query. This means, that if a worker were to update just after the primary process queried the update table, then it would have to wait longer until the next query which would occur after the current query updates had been applied.

This is reflective in the maximum wait times of the workers. The minimum wait time occurs when the primary process recognizes and implements that workers update first. Since this operation is synchronous and intermittent between worker update queries, this causes the variance to grow larger. The worst case scenario would be a worker having to wait for up to six updates before having theirs processed due to missing a query involving the other three workers and then having their update recognized at the bottom of the following query.

A portion of this waiting is actually the primary process retrieving gradients, updating and pushing the global model weights. This is observed in the small variance between timing records for reading gradients and pushing weights from Tables 17 and 19. These two functionalities are synchronous and the combined timing effects the workers wait time heavily.

Each epoch takes around twice as long as the SN architecture, which is seen in Table 18, and varies significantly over the 30 epochs as per the extremely large variance value. All the timing effects discussed previously accumulate into this epoch completion time. There are multiple reads of the global model by workers, pushing gradients, and pushing weights that all accumulate into a larger overhead than the SN architecture. This architecture was performed using four workers with capabilities to scale beyond that. However, there would be a limit as the primary process would become more of a bottleneck than it currently is since workers must wait for their global model updates to complete. Again, like the SN architecture, at the end of each epoch each worker saves their recently pulled model to disk as a measure of redundancy in case there is a system failure. The reason that the PSv architecture takes half as long to save the respective model is the missing optimizing state dictionary that the SN architecture retains.

Future work would be best served investigating efficient mechanisms to asynchronously have workers push gradients, primary update weights, primary push weights, and workers read weights on a layer-by-layer basis. Eliminating the bottleneck of a centralized primary node may force considerations towards ring-AllReduce approaches for gradient sharing. Other approaches for efficiency increase could lie at the gradient accumulation stage. Determining the effects and overcoming these issues at larger accumulation rounds could increase the performance of epoch completion. However, it is unknown how these larger accumulation rounds would affect convergence given the current experiments.

The cost of running deep learning training for the SN architecture is relatively cheap. Especially for early stopped training runs or runs that do not require such complex data. The PSv approach cost over twice as much as the SN architecture within the same amount of time of training. This is obvious since the PSv architecture used more resources. Even using 5 virtual machines to run a distributed parameter server (variant), it is an affordable approach for distributed deep learning training given that the application justifies monetary expense.

Communication bottlenecks are a major concern for the PSv architecture. Overcoming the primary process bottleneck would require a tiered server architecture to overcome and therefore introducing complexity to the system. This involves having workers subscribe to a gradient accumulation server, which would then push to a higher level and pull down the available global model weights. This architecture may introduce a higher degree of weight staleness but could be considered a systematically efficient improvement over the current PSv architecture. Investigations into the effect of stale weights and object detection could be considered for future work. Reducing systematic overhead within the PSv architecture would improve its efficiency in order to exceed standard training times. Within this same vein, implementing an asynchronous task loader communicating with the data server would be very beneficial in regard to architectural performance. However, the biggest bottlenecks lie at the model reads and uploads of both gradients and weights. Future work would be best served investigating efficient mechanisms to asynchronously have workers push gradients, primary update weights, primary push weights, and workers read weights on a layer-by-layer basis. This could drastically reduce the communication and waiting overhead currently implemented. Eliminating the bottleneck of a centralized primary node may force considerations towards ring-AllReduce approaches for gradient sharing.

Other approaches for efficiency increase could lie at the gradient accumulation stage. Determining the effects and overcoming these issues at larger accumulation rounds could increase the performance of epoch completion. However, it is unknown how these larger accumulation rounds would affect convergence given the current experiments.

Naturally, implementing a training cycle using the entire training dataset would be case for future work with both the SN and PSv architectures to determine their performance over unrestricted time and cost. Since the SN model is able to achieve 92% class accuracy, it would be beneficial to determine the minimum sized dataset to achieve the highest class accuracy score. The minimum sized dataset would achieve time, and computation and financial expenses to train a successful model. This information could give insight into similar satellite optical imagery datasets focused on object detection.