# NavP: Enabling Navigational Programming for Science Data Processing via
Application-Initiated Checkpointing

Lei Pan
Jet Propulsion Laboratory
California Institute of Technology
Pasadena, USA
<EMAIL_ADDRESS>
Twinkle Jain
Northeastern University
Boston, USA
<EMAIL_ADDRESS>This work was partially supported by National Science
Foundation Grant OAC-1740218 and a grant from Intel Corporation.

###### Abstract

Science Data Systems (SDS) handle science data from acquisition through
processing to distribution. They are deployed in the Cloud today, and the
efficiency of Cloud instance utilization is critical to success. Conventional
SDS are unable to take advantage of a cost-effective Amazon EC2 spot market,
especially for long-running tasks. Some of the difficulties found in current
practice at NASA/JPL are: a lack of mechanism for app programmers to save
valuable partial results for future processing continuation, the heavy weight
from using container-based (Singularity) sandboxes with more than 200,000 OS-
level files; and the gap between scientists developing algorithms/programs on
a laptop and the SDS experts deploying software in Cloud computing or
supercomputing.

All of the above difficulties can be ameliorated if the process carrying out
computations on the scientist’s laptop can be directly migrated among multiple
nodes in the Cloud or on the supercomputer. The key idea is to allow the
scientist programmers to directly navigate the computations to the data, or
help them move the remaining computations to a new Cloud instance when the old
one is taken away. We present a first proof-of-principle of this using NavP
(Navigational Programming) and fault-tolerant computing (FTC) in SDS, by
employing program state migration facilitated by Checkpoint-Restart (C/R).
NavP provides a new navigational view of computations in a distributed world
for the application programmers. The tool of DHP (DMTCP Hop and Publish) we
developed enables the application programmers to navigate the computation
among instances or nodes by inserting $\mathtt{hop(destination)}$ statements
in their app code, and choose when to publish partial results at stages of
their algorithms that they think worthwhile for future continuation. The
result of using DHP is that a parallel distributed SDS becomes easier to
program and deploy, and this enables more efficient leveraging of the Amazon
EC2 Spot market. This technical report describes a high-level design and an
initial implementation.

NASA/Jet Propulsion Laboratory acquires massive amounts of observatory
instrument data in their missions from planetary (e.g., Mars Exploration,
Cassini, Rosetta, Psyche) to earth science (e.g., Jason, OCO, NISAR, SMAP,
MAIA). This instrument data is then processed into science data on massively
parallel distributed computing facilities such as the Cloud and
supercomputers. Innovations in SDS programming are the key to the success of
JPL missions.

##  1 Introduction

Large-scale Science Data Systems (SDS) like Hybrid Cloud Science Data System
(HySDS) [1] are widely used in various critical NASA missions [1]. Cloud
computing is elastic and scalable, making it an ideal choice for SDS. Another
platform for SDS is cluster computers, an example of which is NASA’s petascale
supercomputer, Pleiades, which has seen heavy investment over the last decade.

Our first goal is to help the scientist programmers to go directly from
algorithms on paper to final production runs in SDS, without much help from an
outside specialist in SDS. By allowing the scientist-programmers to directly
program and immediately test and deploy (without much help of the SDS expert),
the edit-test-debug-production cycle of software development will be greatly
accelerated. Another goal is to facilitate high performance and effective
resource leveraging.

However, in the currently available SDS, there are three major problems that
prevent us from achieving this goal.

  1. 1.

Long-running tasks are not readily broken down into smaller ones to leverage
the Amazon EC2 spot market. This makes it hard or impossible to exploit the
steep discounts of the EC2 spot market. It is the temporal aspect of atomic
jobs.

  2. 2.

The prevailing container-based approach to deploying programs and moving
computation to data is heavyweight, in that it moves much more loaded data
than the strictly necessary program state. In the case of a Singularity
sandbox, which is the virtualization technique adopted on Pleiades, the run-
time environment being moved includes 200,000+ OS-level files. Furthermore,
container images, in which app programs are installed, take a long time to
build, and require knowledge and skills in SDS deployment, such as Terraform
scripting, AWS instance management, Docker/Singularity build scripting, and
Jenkins pipeline scripting, which scientist programmers in general are not
familiar with. One would always have to work with an SDS expert in
virtualizing and deploying the application-level programs, and that is even
true for doing version updates. Part of the problem is mixing different levels
of abstraction for different concerns, such as app algorithms vs. details of
data distribution. Therefore, unnecessary burdens are placed on the scientist-
programmer’s shoulders. This is the spatial aspect of atomic jobs.

  3. 3.

Only embarrassingly parallel algorithms (e.g., MapReduce) are easily
programmable. In general, parallel programming is a non-trivial task,
especially in a distributed environment. This is due to the fact that
prevailing programming methodologies provide a stationary view of the
distributed systems where stationary processes work with each other using
messages. Using an analogy in traveling as an example, there are two views,
namely arrivals-and-departures in the train stations/airports and an itinerary
in a traveler’s hand. These two views both capture the exact same information,
spatial and temporal, about a traveler. Our proposal is that, just as a
traveler should be using an itinerary to travel with, distributed parallel
programming should be using the navigational view. The reality though is that
all prevailing programming paradigms today are as awkward as traveling using
the arrivals-and-departures view. Our proposal comes with a grand challenge:
how to make NavP almost as efficient as message passing, for all major
programming languages. This would mean optimizations at the underlying DMTCP
level: (1) To travel to a remote node carrying only the data needed for future
computation; and (2) To avoid moving code, including both app level code and
run-time environment level code, e.g., OS code, python modules, shared
libraries, to visit a node more than once.

We leverage checkpoint-restart (C/R) to enable Navigational Programming (NavP)
[2] in order to facilitate high performance, effective resource leveraging,
and ease of use for scientist programmers.

Navigational Programming (NavP) was originally developed in 2004 [2]. The
advantage of NavP for distributed parallel programming is that it brings the
computation to the data using an intuitive navigational view of the
distributed programming environment, as opposed to the conventional view of
message passing, which in general would force massive restructuring of the
original sequential algorithms. NavP has not yet been widely adopted because
process migration for multiple programming languages across the board is
difficult. Furthermore, making process migration as efficient as message
passing is a grand challenge at the underlying system level. However, we argue
that the time has now come for this novel paradigm. This work presents an
architecture that leverages a principled implementation relying on
checkpointing in order to enable NavP.

The next five items each present background on the practical requirements of
Scientific Data Systems (SDS), followed by a question that motivates the goals
of the NavP project. This paper answers those questions in the positive, by
showing a first proof-of-principle for NavP.

  1. 1.

Large-scale software development typically uses Continuous Integration (CI)
and Continuous Deployment (CD) to check in, build, test, and deploy
incremental changes in code development. In practice, new versions of the code
are built in and deployed using Docker/Singularity images. The overhead of
this approach is the virtual run-time environment, i.e., some 200,000 files
from the operating system, and cannot be ignored since CI/CD happens
continuously and frequently as scientists update their algorithms. This
overhead can become overwhelming in some cases. For example, NASA’s
supercomputer Pleiades uses Singularity instead of Docker for security
considerations. A Singularity sandbox with more than 200,000 small files will
take an enormous amount of time to handle (untarring or deleting) on Pleiades’
Lustre file system [3].

Can programs be made to migrate themselves? If so, the virtual containers are
then deployed only once, and the subsequent CI/CD operations deploy only the
software applications.

  2. 2.

Scientific application developers are usually scientists who write code on
their laptops or desktops, where data is a small sample, and where processing
power is limited. Going to the Cloud or supercomputer involves CI/CD
administered by SDS experts. So if a scientist develops a new algorithm that
works on laptops with small input data, it will require days or even weeks for
the scientist to try it out on a larger scale. This is not acceptable.

Can we convert the Cloud or supercomputer into a virtual environment that acts
as an extension of the scientist’s laptop? If so, the scientist can test their
new algorithms at any time, even at midnight when a new idea suddenly comes to
them. The program’s self-migration envisioned here enables the scientists to
do just this.

  3. 3.

Webification tools such as JPL’s Pomegranate [4] use the http protocol and web
browsers to open HDF/netCDF files remotely and transfer only what is needed by
the clients. This avoids downloading entire files in cases where only a
fraction is useful. Over the years, people have developed HDF/netCDF readers
that work on local files efficiently.

Can we can migrate the readers to where the HDF/netCDF files are? If so, file
access can be made just as efficient as using HDF/netCDF readers. And at the
same time, all of the existing code can be reused. If this can be done, there
would be no need to develop or learn to use new tools such as Pomegranate.

  4. 4.

Over the past decades, people spent tremendous effort developing techniques
for web services. Corba, JWS, Flask are among the tools provided. These use
data in some format, such as xml or json to facilitate communication between
the client and the server.

Can programs migrate across the server and client run-time environments? If
so, then the right way to facilitate remote communication will be through the
familiar program variables, and programmers will not need to learn anything
new.

  5. 5.

In physics there are two different views of observing moving phenomena: the
Eulerian and the Lagrangian views. These correspond to the arrivals-and-
departures (Eulerian) information shown on the screens of airports or train
stations, versus the itinerary (Lagrangian) held in the travellers’ hand.
Computers were originally invented without networking. Hence processes
typically are not migrated through the network. So programming using the
Eulerian view has been the natural view in programming: all program lines
describe local, stationary computation activities, and when a remote computer
service is needed, message passing or its variant is used. In earlier work,
process self-navigation is proposed as the solution to describe distributed
computation — following its locus of computation in the Lagrangian view [5].
The result is the Distributed Sequential Computing (DSC) program model from
2001 [6]. And multiple DSC’s are then synchronized to carry out parallel
computation using the idea of a Mobile Pipeline (2005) [7]. Scalable
performance and easy of programming has been demonstrated in this way, for
notoriously hard-to-parallelize numerical algorithms.

In the modern networked world, can NavP now be made a first-class object at
the operating system level? Or can NavP use a development toolkit, such as
DMTCP, which in theory can be done as efficiently as message passing, by
switching between the two different views? This would be used to describe
exactly the same physical phenomenon, using the same amount of message
passing, along with some small implementation overhead.

##  2 Background

###  2.1 SDS

Science Data Systems (SDS) handle science data from acquisition through
processing to distribution. This makes a generic SDS a multi-stage workflow,
as depicted in Figure 1. Science values are being put into the data as higher
level processing are being advanced to. Typically, high level processing
requires capabilities involving petascale data storage and processing power.

Figure 1: Data processing phases in SDS

###  2.2 Fault-Tolerant Computing in the Cloud

Amazon provides EC2 spot instances, which offer spare compute capacity in the
AWS Cloud, available at steep discounts (90% savings), but they can be taken
away at any time. In the meantime, each atomic task can take hours to finish.
Our strategy is to break the original task into smaller pieces using
checkpointing and introduce Fault-Tolerant Computing. So, the “remaining”
computation can be brought to and restarted on a new instance after the old
instance disappears.

###  2.3 NavP: Navigational Programming

A distributed parallel system is not directly programmable by scientist-
programmers. One would always have to work with an SDS expert in virtualizing
and deploying the application-level programs, and even when doing version
updates. The levels of abstraction for different concerns, application
algorithms vs. details of distribution, are coupled. Scientists develop
original algorithms that carry out numerical calculations toward data in the
form of abstract data structures and variables. The reality of petascale
distributed data over the networked systems forces the scientists to work
closely with SDS experts to restructure their original algorithms into actual
code (e.g., client and server code), install the code in virtual boxes (e.g.,
Docker images), and deploy the programs in virtual containers onto computer
nodes or Cloud instances. Therefore unnecessary burdens are placed on the
application developers' shoulders. For applications that are not by nature
embarrassingly parallel, this task is difficult, if not impossible. NavP was
introduced to address these difficulties [2]. A new view of distributed
programming, namely the NavP view, is introduced. In this view, the
description of a computation follows its locus to where the large data is
found [5]. This is done by inserting $\mathtt{hop()}$ statements in the
original sequential code. A $\mathtt{hop(dest)}$ statement pauses the
computation, collects all the program/thread state, migrates to the
$\mathtt{dest}$ node, and resumes computation.

###  2.4 Application-initiated transparent C/R

We use the DMTCP (Distributed MultiThreaded CheckPointing) [8] package for
C/R. It transparently checkpoints computations in user-space. It saves a copy
of the program state (called a Checkpoint Memory Image (CMI)) to disk and
resumes the process later. It requires no modifications to the user
application code nor to the operating system.

The DMTCP plugins [9] and hooks for application-initiated checkpointing
provide a flexible way to introduce add-on behaviors around DMTCP events [8].

We develop DHP (DMTCP Hop and Publish), a new Python tool, around DMTCP. This
provides two utilities around the checkpoint and restart: a) hop(dest) — i.e.,
responsible for generating CMI on the source node, migrating to, and resuming
CMI on the destination node; and b) publish(dest, status) — i.e., responsible
of publishing job’s status and result on the destination node.

The next section describes how to enable $\mathtt{hop()}$ and FTC using DHP
and web services for program state migration.

##  3 The NavP Bridging Services (NBS)

###  3.1 The NavP Bridging Services (NBS) with DHP

The NavP Bridging Services (NBS) run on each compute node and serve the client
requests from the application processes. The NBS service for communication and
data migration (i.e., $\mathtt{svc/hop}$), running on both source and
destination nodes, enables DHP to migrate CMI(s) between two nodes. We provide
an overview of the workflow of a NBS in Figure 2. In the figure, there is a
source node i.e., $\mathtt{Host~{}A}$ and a destination node i.e.,
$\mathtt{Host~{}B}$ with NBS services running on each node. Now, an
application process running on $\mathtt{Host~{}A}$ can call the
$\mathtt{DHP.hop(B)}$ utility, which in turn initiates checkpoint that
generates CMI(s) as well as a restart script, and then calls the
$\mathtt{svc/hop}$ NBS service running on $\mathtt{Host~{}B}$ before it
terminates itself. The $\mathtt{svc/hop}$ NBS service on $\mathtt{Host~{}B}$
copies the CMI and restart script from $\mathtt{Host~{}A}$ to $\mathtt{B}$,
and runs the restart script to resume the computation. This way DHP is able to
utilize existing data migration utility.

Figure 2: Overview of NavP Bridging Service workflow

###  3.2 NBS to enable NavP

We use $\mathtt{DHP.hop(dest)}$ and $\mathtt{svc/hop}$ to enable NavP.
Pseudocode for the $\mathtt{DHP.hop(dest)}$ is in Figure 3, in which S3 means
some shared disk volume, either in an S3 bucket or bound to the containers.
Pseudocode for $\mathtt{svc/hop}$ is in Figure 4.

(1) $\mathtt{checkpoint()}$ (2) $\mathtt{if\ isResume():\\#\ after\
checkpointing}$ (3) $\mathtt{\ \ \ copy\ CMI\ and\ restart\ script\ to\ S3\ }$
(4) $\mathtt{\ \ \ request\ svc/hop\ on\ dest\ }$ (5) $\mathtt{\ \ \ exit\ }$

Figure 3: Pseudocode for the $\mathtt{DHP.hop(dest)}$ utility

(1) $\mathtt{copy\ CMI\ and\ restart\ script\ from\ S3\ }$ (2) $\mathtt{run\
dmtcp\\_restart\\_script.sh\ }$

Figure 4: Pseudocode for the $\mathtt{svc/hop}$ service

###  3.3 NBS to enable FTC in SDS

When jobs in SDS are treated as atomic operations, they can be either “new”
before the run, or “finished” after the run. Any interrupted jobs return to
the “new” status. A new job has input datasets, and a finished job has
products. For the sake of brevity, we ignore more sophisticated situations,
such as a “running” status, in which a job can have both input datasets and
partial products. These additional combinations can be handled in real life
applications using the same principle.

The key idea is to introduce a new job status, called “ckpt”, in which the CMI
is treated as a “special product”. We implement three services to handle the
following jobs: (1) $\mathtt{svc/list\\_jobs}$: this returns all jobs with
their $\mathtt{job\\_id^{\prime}s}$ along with their statuses, such as those
shown in Figure 5.

(1) $\mathtt{[[``1",``new"],}$ (2) $\mathtt{\>[``2",``ckpt"],}$ (3)
$\mathtt{\>[``3",``finished"]]}$

Figure 5: Sample list of jobs with job-ID and status

(2) $\mathtt{svc/get\\_job}$: this returns the status of a job given its
$\mathtt{job\\_id}$, or the next job that is not finished when no
$\mathtt{job\\_id}$ is provided. (3) $\mathtt{svc/publish\\_job}$: this
publishes jobs with two possible statuses: “ckpt”, in which case the CMI and
restart script are uploaded, and “finished”, in which case the final product
is uploaded.

A DHP utility, $\mathtt{DHP.publish(dest,status)}$, is listed in Figure 6.
This is similar to $\mathtt{DHP.hop(dest)}$. An application uses
$\mathtt{DHP.publish()}$ to publish with two possible statuses: “ckpt”
(checkpoints and calls $\mathtt{svc/publish\\_job}$ service with a “ckpt”
status) or “finished” (calls $\mathtt{svc/publish\\_job}$ with a “finished”
status). The “dest” is the job scheduler service.

(1) $\mathtt{if\ status==``ckpt":}$ (2) $\mathtt{\ \ checkpoint()}$ (3)
$\mathtt{\ \ if\ isResume():\\#\ after\ checkpointing}$ (4) $\mathtt{\ \ \
copy\ CMI\ and\ restart\ script\ to\ S3\ }$ (5) $\mathtt{\ \ \ request\
svc/publish(dest,``ckpt")\ }$ (6) $\mathtt{elseif\ status==``finished":}$ (7)
$\mathtt{\ \ \ copy\ product\ to\ S3\ }$ (8) $\mathtt{\ \ \ request\
svc/publish(dest,``finished")\ }$

Figure 6: Pseudocode for $\mathtt{DHP.publish(dest,status)}$ utility

##  4 Experiments: Proof of Concept

Our experiments are designed to answer the following questions:

  1. 1.

How to enable NavP in SDS?

The test case application chosen is called “the co-location of satellite
observation data”. Specifically, the data from the instrument VIIRS (Visible
Infrared Imaging Radiometer Suite) is mapped to the geometry of CrIS (Cross-
track Infrared Sounder) [10, 11, 12]. The first experiment has the pseudocode
listed in Figure 7, in which the calls to svc/publish(‘‘ckpt") are where we
checkpoint and where the CMIs are published as “partial products”. So when the
execution fails for any reason at any time, restart will happen from the most-
recent-checkpoint. The application programmer is thus in control of where and
how frequently checkpoints happen. The last call to svc/publish(‘‘finished")
publishes the real product.

(1) $\mathtt{request\ svc/get\\_job\ to\ get\ job\\_id/status\ }$ (2)
$\mathtt{if\ status==``new":}$ (3) $\mathtt{\ \ main(job\\_id)}$ (4)
$\mathtt{elseif\ status==``ckpt":}$ (5) $\mathtt{\ \ DHP.restart(job\\_id)}$
(6) $\mathtt{def\ main(job\\_id):}$ (7) $\mathtt{\ \ read\ VIIRS\ data}$ (8)
$\mathtt{\ \ read\ CrIS\ data}$ (9) $\mathtt{\ \ DHP.publish(job\\_id,\
``ckpt")}$ (10) $\mathtt{\ \ compute\ CrIS\ LOS\ vector\ in\ ECEF}$ (11)
$\mathtt{\ \ compute\ VIIRS\ POS\ vector\ in\ ECEF}$ (12) $\mathtt{\ \
DHP.publish(job\\_id,\ ``ckpt")}$ (13) $\mathtt{\ \ match\ VIIRS\ to\ CrIS}$
(14) $\mathtt{\ \ write\ product}$ (15) $\mathtt{\ \ DHP.publish(job\\_id,\
``finished")}$

Figure 7: Pseudocode for JPL application of VIIRS/CrIS co-location.

The second experiment assumes that the input data granules are available on a
remote server. So the application needs to first hop to read the input data.
Assuming further that this remote server is only for hosting data (input data
and output product), the application hops back to the client server to carry
out co-location matching, before hopping to the remote server again to publish
the final product. The pseudocode listed in Figure 8 therefore has three
$\mathtt{hop()}$ statements inserted in the original code.

(1) $\mathtt{DHP.hop(other)}$ (2) $\mathtt{read\ VIIRS\ data}$ (3)
$\mathtt{read\ CrIS\ data}$ (4) $\mathtt{DHP.hop(other)}$ (5)
$\mathtt{compute\ CrIS\ LOS\ vector\ in\ ECEF}$ (6) $\mathtt{compute\ VIIRS\
POS\ vector\ in\ ECEF}$ (7) $\mathtt{match\ VIIRS\ to\ CrIS}$ (8)
$\mathtt{DHP.hop(other)}$ (9) $\mathtt{write\ product}$

Figure 8: Pseudocode for NavP program of the JPL application of VIIRS/CrIS co-
location.

  2. 2.

What is the performance overhead of DMTCP checkpointing and restart? How does
this overhead, which is primarily incurred by disk I/O, from writing out and
loading up the CMIs, as compared to the cost of migrating the CMIs over the
network?

There are two experimental environments, chosen for two different purposes.
The first is a single desktop computer running the Linux operating system,
with an Intel Xeon 4GHz CPU with 8 cores and 32GB of physical memory. Two
virtual NBS containers are run on the same computer, making process migration
to happen virtually remotely, but physically only locally. This removes
networking cost from the equation, making it ideal for assessing the cost and
impact of CMI from checkpointing done by DMTCP. The second system is the
Amazon Cloud with m4.4xlarge instances and AWS S3 buckets for storage. This is
the real world for JPL SDS, in which the network and S3 performance are all
taken into consideration.

We have implementations of all the pseudocode listed in this report working.
From this, we quickly realized that our general purpose DMTCP needs to be
optimized before further experiments can be meaningfully carried out. This is
because as a general purpose C/R tool, DMTCP currently puts everything our app
needs from the run-time environment, e.g., python modules and shared
libraries, into the CMIs. As a result, the cost of disk I/O and network
transfer of CMIs overshadows the cost of numerical computation in the app. The
Docker/Singularity containers to which the app does hop() or publish(), are
assumed to have identical run-time environments, with the same
python/modules/shared libraries etc. So DMTCP can conveniently skip
checkpointing anything related to the run-time environment. Upon arriving at a
remote container, the DMTCP restart shell script will load the modules and
shared libraries from the local installation. This will make the CMIs much
more lightweight than they currently are.

Hence, our immediate future work includes a special-purpose, NavP-oriented
DMTCP that will be optimized for navigational programming.

##  5 Discussion

In this section, we answer following questions related to the current work.

Question 1. How to predict when to checkpoint considering spot instance can
take their resources back anytime?

Amazon EC2 spot instances are cost-effective resources but can be reclaimed
with just two minutes of instance termination notice. This is not sufficient
time to initiate, complete, and migrate CMIs for a large-scale job. Jangjaimon
et. al [13] takes account of past resource reclaim events to predict a
checkpoint interval for their adaptive checkpoint scheme. However, the result
of an incomplete job is equivalent to no result for an SDS job. Therefore,
prediction of the next resource reclaim event would not benefit the user
significantly.

Question 2. What are some advantages of application-initiated checkpoint over
conventional periodic checkpoint?

Fixed-interval periodic checkpoints work well where jobs are not considered
atomic, and where intermediate results can also be useful. However,
application-initiated checkpoint is an application-aware technique and enables
two important features: 1) applications that interact with large data often
have a small memory footprint before and after the job, which makes the
checkpointing task faster, and makes the CMI’s size feasible for migration
over the network; and 2) more control and awareness by the application where
it’s safe to checkpoint with a minimal side effect.

Question 3. How to keep a CMI’s size small?

We either migrate CMIs over the network or write on a shared disk for NavP.
Amazon S3 buckets are often not the fastest I/O option because of their
physical location. Therefore, it is important to keep CMIs small in size. The
size directly depends on the memory footprint (and associated files if one
wants to checkpoint open files as well). One solution is to make the memory
footprint small by zeroing out or unmapping unused memory within process
address space at the time of checkpoint. However, this would require changes
in the application, and the solution would no longer remain transparent.

Another solution is to save the CMIs incrementally by saving only deltas of
each consecutive checkpoint. This would optimize I/O interaction, but this
would require an extra step to replay deltas at restart with the current DMTCP
open-source code. We checkpoint after a job finishes and before the
application loads another data set into memory, in order to ensure a small
size for the CMI. DHP currently replaces the last CMI with the latest CMI. In
addition, incremental checkpointing is an option for future work.

Question 4. What if the checkpoint task is interrupted?

DHP guarantees an atomic checkpointing phase. Thus, DHP makes sure to not
replace previous CMIs if the resources were reclaimed in the middle of an
ongoing checkpointing phase.

Question 5. What are some future optimization to this work?

Currently, DHP writes a CMI first on the shared disk, and then it migrates the
saved CMI over network. With a small modification to DMTCP, it is possible to
directly stream CMIs over the network, in a manner similar to live migration.
Also, currently, checkpoint and migration is a two-step process. However,
consider a scenario in which the checkpointing task is finished, but the
migration has yet to finish. This can lead to an inconsistent state.
Therefore, we currently consider DHP.hop to be complete only once the
migration is complete.

Question 6. What is the difference between existing adaptive checkpoint and
the current work?

The current implementation doesn’t take account of previous events, duration,
and patterns of EC2 spot instance termination. However, this information can
help in choosing a candidate node where DHP should migrate CMIs to within
cloud. One can choose a node that is unlikely to be terminated before a job
finishes.

##  6 Related Work

Fault tolerance for long running application in high performance computing
(HPC) has been a major area for checkpoint-restart. In recent years,
checkpointing is found useful for cost-effective resource utilization within
and beyond HPC. Jangjaimon et al. [13] discuss effective cost reduction for
elastic clouds through adaptive checkpointing. However, the current work does
not rely on previous events to checkpoint but focuses on keeping the CMI size
small to allow checkpointing at any time.

Yi et al. [14] compares several checkpointing schemes including adaptive
checkpointing based on hourly, rising edge, basic adaptive and current-price
events. However, they assume that the checkpointing cost is known, which is
unlikely when the memory footprint varies throughout the workflow. In
addition, atomic jobs are not flexible enough to be checkpointed at an
arbitrary time. This adds an extra restriction to the existing checkpointing
schemes for applications in the cloud.

In order to make checkpoint invocation cost-effective, Jung et al. [15]
predicts termination (out-of-bid) and a price hike for spot instances, based
on the history. They use Virtual Machine-based (VM-based) checkpoint and
migration to avoid any waiting time at restart. However, migrating a VM
instead of an application requires more data transfer over the network, and
the migration cost varies based on the network. The current work focuses on
reducing the memory footprint to minimize data transfer, while migrating CMIs
as data.

To the best of our knowledge, this is the first work to enable the
navigational programming model for SDS applications via checkpoint-restart
over commercial cloud systems, such as Amazon EC2 spot market.

##  7 Conclusion and Future Work

This is an initial attempt in applying Navigational Programming, enabled with
process checkpointing by DMTCP, to NASA/JPL science data processing. We have
successfully built the NavP Bridging Services (NBS) and demonstrated its
application for JPL Satellite Data Co-location processing. With our current
implementation, we are able to successfully checkpoint, migrate, and resume
our app. We also mimicked successfully the continuation of a job on a new
instance with intermediate partial results published during the most recent
checkpointing. Initial performance studies showed large overhead in our
current implementation. We plan to customize and optimize DMTCP, and publish
case studies and performance analysis, in a future version of this report.

This report introduced and described several concepts in the NavP paradigm,
and provided a rationale behind why this is superior to the currently
prevailing methods. It has also raised many questions. More case studies are
necessary to provide experimental evidence. With an optimized NBS/DMTCP system
in the future, we plan to attack more JPL science data processing problems,
and answer the questions that were raised in this report.

## Acknowledgment

The authors wish to acknowledge useful discussions on this topic with Gene
Cooperman.

## References

  * [1] H. Hua, G. Manipon, and the JPL HySDS Team, “The hybrid cloud science data system,” https://hysds-core.atlassian.net/wiki/spaces/HYS/overview, Sep. 2021.
  * [2] L. Pan, M. K. Lai, K. Noguchi, J. J. Huseynov, L. F. Bic, and M. B. Dillencourt, “Distributed parallel computing using navigational programming,” in _International Journal of Parallel Programming_ , vol. 32, no. 1. Springer, Feb 2004.
  * [3] X. Li, B. Dong, L. Xiao, L. Ruan, and Y. Ding, “Small files problem in parallel file system,” in _2011 International Conference on Network Computing and Information Security_ , vol. 2, 2011, pp. 227–232.
  * [4] Z. Xing, “Webification for earth science,” Jet Propulsion Laboratory, California Institute of Technology, Pasadena, CA, Tech. Rep., Mar 2018, white Paper.
  * [5] L. Pan, L. F. Bic, M. B. Dillencourt, and M. K. Lai, “NavP versus SPMD: Two views of distributed computation,” in _Proceedings of the Fifteenth IASTED International Conference on Parallel and Distributed Computing and Systems_ , T. Gonzalez, Ed., vol. 2, Algorithms. Anaheim, Calif.: ACTA Press, Nov. 2003, pp. 666–673.
  * [6] L. Pan, L. Bic, and M. Dillencourt, “Distributed sequential computing using mobile code: Moving computation to data,” in _International Conference on Parallel Processing, 2001._ , 2001, pp. 77–84.
  * [7] L. Pan, M. K. Lai, M. B. Dillencourt, and L. F. Bic, “Mobile Pipelines: Parallelizing left-looking algorithms using Navigational Programming,” in _High Performance Computing – HiPC 2005_ , D. A. Bader, M. Parashar, V. Sridhar, and V. K. Prasanna, Eds. Berlin, Heidelberg: Springer Berlin Heidelberg, 2005, pp. 201–212.
  * [8] J. Ansel, K. Arya, and G. Cooperman, “DMTCP: Transparent checkpointing for cluster computations and the desktop,” in _2009 IEEE International Symposium on Parallel & Distributed Processing_. IEEE, 2009, pp. 1–12.
  * [9] G. C. Kapil Arya, “Dmtcp: Bringing checkpoint-restart to python,” in _Test Proceedings for OpenProceedings_ , A. Zonca, Ed., 2013. [Online]. Available: http://dx.doi.org/10.6084/m9.figshare.935502
  * [10] L. Wang, D. Tremblay, B. Zhang, and Y. Han, “Fast and accurate collocation of the visible infrared imaging radiometer suite measurements with cross-track infrared sounder,” _Remote Sensing_ , vol. 8, no. 1, 2016. [Online]. Available: https://www.mdpi.com/2072-4292/8/1/76
  * [11] Q. Yue, E. J. Fetzer, L. Wang, B. H. Kahn, I. Tkatcheva, M. Schreier, B. Lambrigsten, N. Smith, J. Blaisdell, and K. G. Meyer, “Evaluating the consistency and continuity of cloud property data records from aqua and snpp on the pixel scale,” submitted to Journal of Geophysical Research: Atmospheres.
  * [12] E. J. Fetzer, Q. Yue, S. Wong, A. Guillaume, R. A. Stachnik, T. Wang, G. Manipon, B. H. Kahn, B. D. Wilson, and H. Hua, “A multi-sensor water vapor, temperature and cloud climate data record from the A-train,” in _American Geophysical Union, Fall Meeting 2016_ , Dec 2016.
  * [13] I. Jangjaimon and N.-F. Tzeng, “Effective cost reduction for elastic clouds under spot instance pricing through adaptive checkpointing,” _IEEE Transactions on Computers_ , vol. 64, no. 2, pp. 396–409, 2013.
  * [14] S. Yi, A. Andrzejak, and D. Kondo, “Monetary cost-aware checkpointing and migration on Amazon cloud spot instances,” _IEEE Transactions on Services Computing_ , vol. 5, no. 4, pp. 512–524, 2011.
  * [15] D. Jung, S. Chin, K. S. Chung, and H. Yu, “VM migration for fault tolerance in spot instance based cloud computing,” in _International Conference on Grid and Pervasive Computing_. Springer, 2013, pp. 142–151.