We present POSH, a framework that accelerates shell applications with I/O-heavy components, such as data analytics with command-line utilities. Remote storage such as networked filesystems can severely limit the performance of these applications: data makes a round trip over the network for relatively little computation at the client. Reducing the data movement by moving the code to the data can improve performance.
POSH automatically optimizes unmodified I/O-intensive shell applications running over remote storage by offloading the I/O-intensive portions to proxy servers closer to the data. A proxy can run directly on a storage server, or on a machine closer to the storage layer than the client. POSH intercepts shell pipelines and uses metadata called annotations to decide where to run each command within the pipeline. We address three principal challenges that arise: an annotation language that allows POSH to understand which files a command will access, a scheduling algorithm that places commands to minimize data movement, and a system runtime to execute a distributed schedule but retain local semantics.
We benchmark POSH on real shell pipelines such as image processing, network security analysis, log analysis, distributed system debugging, and git. We find that POSH provides speedups ranging from 1.6× to 15× compared to NFS, without requiring any modifications to the applications. more »« less
Wu, Chenggang; Sreekanti, Vikram; Hellerstein, Joseph M.(
, SIGMOD '20: Proceedings of the 2020 ACM SIGMOD International Conference on Management of Data)
null
(Ed.)
We consider the setting of serverless Function-as-a-Service (FaaS) platforms, where storage services are disaggregated from the machines that support function execution. FaaS applications consist of compositions of functions, each of which may run on a separate machine and access remote storage. The challenge we address is improving I/O latency in this setting while also providing application-wide consistency. Previous work has explored providing causal consistency for individual I/Os by carefully managing the versions stored in a client-side data cache. In our setting, a single application may execute multiple functions across different nodes, and therefore issue interrelated I/Os to multiple distinct caches. This raises the challenge of Multisite Transactional Causal Consistency (MTCC): the ability to provide causal consistency for all I/Os within a given transaction even if it runs across multiple physical sites. We present protocols for MTCC implemented in a system called HYDROCACHE. Our evaluation demonstrates orders-of-magnitude performance improvements due to caching, while also protecting against consistency anomalies that otherwise arise frequently.
Campbell, C.; Mecca, N.; Duong, T.; Obeid, I.; Picone, J.(
, IEEE Signal Processing in Medicine and Biology Symposium (SPMB))
Obeid, Iyad; Selesnick, Ivan; Picone, Joseph
(Ed.)
The goal of this work was to design a low-cost computing facility that can support the development of an
open source digital pathology corpus containing 1M images [1]. A single image from a clinical-grade digital
pathology scanner can range in size from hundreds of megabytes to five gigabytes. A 1M image database
requires over a petabyte (PB) of disk space. To do meaningful work in this problem space requires a
significant allocation of computing resources. The improvements and expansions to our HPC (highperformance
computing) cluster, known as Neuronix [2], required to support working with digital
pathology fall into two broad categories: computation and storage. To handle the increased computational
burden and increase job throughput, we are using Slurm [3] as our scheduler and resource manager. For
storage, we have designed and implemented a multi-layer filesystem architecture to distribute a filesystem
across multiple machines. These enhancements, which are entirely based on open source software, have
extended the capabilities of our cluster and increased its cost-effectiveness.
Slurm has numerous features that allow it to generalize to a number of different scenarios. Among the most
notable is its support for GPU (graphics processing unit) scheduling. GPUs can offer a tremendous
performance increase in machine learning applications [4] and Slurm’s built-in mechanisms for handling
them was a key factor in making this choice. Slurm has a general resource (GRES) mechanism that can be
used to configure and enable support for resources beyond the ones provided by the traditional HPC
scheduler (e.g. memory, wall-clock time), and GPUs are among the GRES types that can be supported by
Slurm [5]. In addition to being able to track resources, Slurm does strict enforcement of resource allocation.
This becomes very important as the computational demands of the jobs increase, so that they have all the
resources they need, and that they don’t take resources from other jobs. It is a common practice among
GPU-enabled frameworks to query the CUDA runtime library/drivers and iterate over the list of GPUs,
attempting to establish a context on all of them. Slurm is able to affect the hardware discovery process of
these jobs, which enables a number of these jobs to run alongside each other, even if the GPUs are in
exclusive-process mode.
To store large quantities of digital pathology slides, we developed a robust, extensible distributed storage
solution. We utilized a number of open source tools to create a single filesystem, which can be mounted
by any machine on the network. At the lowest layer of abstraction are the hard drives, which were split into
4 60-disk chassis, using 8TB drives. To support these disks, we have two server units, each equipped with
Intel Xeon CPUs and 128GB of RAM. At the filesystem level, we have implemented a multi-layer solution
that: (1) connects the disks together into a single filesystem/mountpoint using the ZFS (Zettabyte File
System) [6], and (2) connects filesystems on multiple machines together to form a single mountpoint using
Gluster [7].
ZFS, initially developed by Sun Microsystems, provides disk-level awareness and a filesystem which takes
advantage of that awareness to provide fault tolerance. At the filesystem level, ZFS protects against data
corruption and the infamous RAID write-hole bug by implementing a journaling scheme (the ZFS intent
log, or ZIL) and copy-on-write functionality. Each machine (1 controller + 2 disk chassis) has its own separate ZFS filesystem. Gluster, essentially a meta-filesystem, takes each of these, and provides the means
to connect them together over the network and using distributed (similar to RAID 0 but without striping
individual files), and mirrored (similar to RAID 1) configurations [8].
By implementing these improvements, it has been possible to expand the storage and computational power
of the Neuronix cluster arbitrarily to support the most computationally-intensive endeavors by scaling
horizontally. We have greatly improved the scalability of the cluster while maintaining its excellent
price/performance ratio [1].
Zhang, Fan; Jin, Tong; Sun, Qian; Romanus, Melissa; Bui, Hoang; Klasky, Scott; Parashar, Manish(
, Concurrency and Computation: Practice and Experience)
Summary
Coupled scientific simulation workflows are composed of heterogeneous component applications that simulate different aspects of the physical phenomena being modeled and that interact and exchange significant volumes of data at runtime. As the data volumes and generation rates keep growing, the traditional disk I/O–based data movement approach becomes cost prohibitive, and workflow requires more scalable and efficient approach to support the data movement. Moreover, the cost of moving large volume of data over system interconnection network becomes dominating and significantly impacts the workflow execution time. Minimize the amount of network data movement and localize data transfers are critical for reducing such cost. To achieve this, workflow task placement should exploit data locality to the extent possible and move computation closer to data. In this paper, we investigate applying in‐memory data staging and data‐centric task placement to reduce the data movement cost in large‐scale coupled simulation workflows. Specifically, we present a distributed data sharing and task execution framework that (1) co‐locates in‐memory data staging on application compute nodes to store data that needs to be shared or exchanged and (2) uses data‐centric task placement to map computations onto processor cores that a large portion of the data exchanges can be performed using the intra‐node shared memory. We also present the implementation of the framework and its experimental evaluation on Titan Cray XK7 petascale supercomputer.
Cloud computing has become a major approach to help reproduce computational experiments. Yet there are still two main difficulties in reproducing batch based big data analytics (including descriptive and predictive analytics) in the cloud. The first is how to automate end-to-end scalable execution of analytics including distributed environment provisioning, analytics pipeline description, parallel execution, and resource termination. The second is that an application developed for one cloud is difficult to be reproduced in another cloud, a.k.a. vendor lock-in problem. To tackle these problems, we leverage serverless computing and containerization techniques for automated scalable execution and reproducibility, and utilize the adapter design pattern to enable application portability and reproducibility across different clouds. We propose and develop an open-source toolkit that supports 1) fully automated end-to-end execution and reproduction via a single command, 2) automated data and configuration storage for each execution, 3) flexible client modes based on user preferences, 4) execution history query, and 5) simple reproduction of existing executions in the same environment or a different environment. We did extensive experiments on both AWS and Azure using four big data analytics applications that run on virtual CPU/GPU clusters. The experiments show our toolkit can achieve good execution performance, scalability, and efficient reproducibility for cloud-based big data analytics.
Anderson, Thomas; Canini, Marco; Kim, Jongyul; Kostic, Dejan; Kwon, Youngjin; Peter, Simon; Reda, Waleed; Schuh, Henry; Witchel, Emmett(
, 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20))
null
(Ed.)
The adoption of low latency persistent memory modules (PMMs) upends the long-established model of remote storage for distributed file systems. Instead, by colocating computation with PMM storage, we can provide applications with much higher IO performance, sub-second application failover, and strong consistency. To demonstrate this, we built the Assise distributed file system, based on a persistent, replicated coherence protocol that manages client-local PMM as a linearizable and crash-recoverable cache between applications and slower (and possibly remote) storage. Assise maximizes locality for all file IO by carrying out IO on process-local, socket-local, and client-local PMM whenever possible. Assise minimizes coherence overhead by maintaining consistency at IO operation granularity, rather than at fixed block sizes.
We compare Assise to Ceph/BlueStore, NFS, and Octopus on a cluster with Intel Optane DC PMMs and SSDs for common cloud applications and benchmarks, such as LevelDB, Postfix, and FileBench. We find that Assise improves write latency up to 22x, throughput up to 56x, fail-over time up to 103x, and scales up to 6x better than its counterparts, while providing stronger consistency semantics.
Raghavan, Deepti, Fouladi, Sadjad, Levis, Philip, and and Zaharia, Matei.
"POSH: A Data-Aware Shell". USENIX ATC 2020 (). Country unknown/Code not available. https://par.nsf.gov/biblio/10213412.
@article{osti_10213412,
place = {Country unknown/Code not available},
title = {POSH: A Data-Aware Shell},
url = {https://par.nsf.gov/biblio/10213412},
abstractNote = {We present POSH, a framework that accelerates shell applications with I/O-heavy components, such as data analytics with command-line utilities. Remote storage such as networked filesystems can severely limit the performance of these applications: data makes a round trip over the network for relatively little computation at the client. Reducing the data movement by moving the code to the data can improve performance. POSH automatically optimizes unmodified I/O-intensive shell applications running over remote storage by offloading the I/O-intensive portions to proxy servers closer to the data. A proxy can run directly on a storage server, or on a machine closer to the storage layer than the client. POSH intercepts shell pipelines and uses metadata called annotations to decide where to run each command within the pipeline. We address three principal challenges that arise: an annotation language that allows POSH to understand which files a command will access, a scheduling algorithm that places commands to minimize data movement, and a system runtime to execute a distributed schedule but retain local semantics. We benchmark POSH on real shell pipelines such as image processing, network security analysis, log analysis, distributed system debugging, and git. We find that POSH provides speedups ranging from 1.6× to 15× compared to NFS, without requiring any modifications to the applications.},
journal = {USENIX ATC 2020},
author = {Raghavan, Deepti and Fouladi, Sadjad and Levis, Philip and and Zaharia, Matei.},
editor = {null}
}
Warning: Leaving National Science Foundation Website
You are now leaving the National Science Foundation website to go to a non-government website.
Website:
NSF takes no responsibility for and exercises no control over the views expressed or the accuracy of
the information contained on this site. Also be aware that NSF's privacy policy does not apply to this site.
