Michael M Lee

Indrajit Roy
email: indrajitr@hp.com

Alvin Auyoung
email: alvina@hp.com

K R Jayaram
email: jayaramkr@hp.com

Yuanyuan Zhou
email: yyzhou@cs.ucsd.edu

Views and Transactional Storage for Large Graphs

Keywords: Graphs, transactions, views, event-driven analysis

A growing number of applications store and analyze graphstructured data. These applications impose challenging infrastructure demands due to a need for scalable, high-throughput, and low-latency graph processing. Existing state-of-the-art storage systems and data processing systems are limited in at least one of these dimensions, and simply layering these technologies is inadequate. We present Concerto, a graph store based on distributed, in-memory data structures. In addition to enabling efficient graph traversals by co-locating graph nodes and associated edges where possible, Concerto provides transactional updates while scaling to hundreds of nodes. Concerto introduces graph views to denote sub-graphs on which user-defined functions can be invoked. Using graph views, programmers can perform event-driven analysis and dynamically optimize application performance. Our results show that Concerto is significantly faster than in-memory MySQL, in-memory Neo4j, and GemFire for graph insertions as well as graph queries. We demonstrate the utility of Concerto's features in the design of two real-world applications: real-time incident impact analysis on a road network and targeted advertising in a social network.

Introduction

Graph-processing applications are quickly emerging as a critical component in domains like social networks, road traffic, and biological networks where data exhibit natural graph structure. Building large-scale graph applications requires middleware support for storing data and for accelerating graph queries. Many of today's graph applications exhibit a need for high volume storage, low-latency updates, and interactive responsiveness. Individually, these requirements do not present a unique challenge, but taken together, they pose a significant challenge to both state-of-the-art storage and data-processing systems. We detail two of these emerging requirements, and how they translate into challenges for the supporting system infrastructure:

-Scalability and consistency. Because many of today's graph applications run on the critical path of an on-line workflow, a graph store needs to provide a combination of adequate query throughput and data ingestion rate. For example, Facebook receives more than 200,000 events per second [START_REF]Facebook's new realtime analytics system: Hbase to process 20 billion events per day[END_REF] while Twitter ingests approximately 80 TB/day of new data [START_REF]Twitter by the numbers[END_REF]. However, without meaningful consistency semantics, such as transactional guarantees, writing distributed graph applications will be difficult and error prone. -Event-driven processing. Some graph applications, such as those used by emergency technicians to respond to incidents, must be real-time to be useful [START_REF] Miller | An empirical analysis of the impact of incidents on freeway traffic[END_REF]. For example, the California highway road sensors requires ingestion of new data every 30 seconds for over 26,000 sensors [START_REF]Caltrans performance measurement system[END_REF]. Such an application is largely event-driven based upon incidents (i.e., accidents, slowdowns) occurring on the road graph, and triggers computation to predict the spread and duration of the incidents. Supporting an API with flexible event processing on the graph store eases the development of these graph applications. Multiple applications monitoring similar events can avoid redundant client-side computation on event detection. Moreover, events can be used to monitor and dynamically optimize the store itself (such as graph layout) to further improve query performance.

Limitations of current systems

State-of-the-art solutions are not designed to address these challenges simultaneously. Traditional storage systems such as relational databases and NoSQL stores do not inherently retain the structure of the graph, and are unsuitable for computing graph algorithms [START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF]. As shown in Table 1, using a fast caching layer such as Memcached on top of a relational database can scale the performance of resolving graph queries. However, this approach relies on pre-computing the set of queries, and thus requires the workload to be known in advance, or returning a computation based upon stale data. Distributed in-memory stores (GemFire [START_REF]GemFire: Technical white paper[END_REF]) provide dynamic scalability and high performance while also supporting transactions. However, similar to traditional relational databases, they do not provide native support for graph objects and hence are slow for graph queries.

Contemporary data-processing frameworks such as MapReduce [START_REF] Dean | MapReduce: Simplified data processing on large clusters[END_REF], Pregel [START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF], Spark [START_REF] Zaharia | Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[END_REF], or GraphLab [START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF] optimize for batch analysis by assuming data is largely read-only, and hence are ill-suited for concurrent read and write queries. Since many of these systems are designed for long running distributed graph analyses, the overhead from additional communication and setup costs may exceed the actual computation time for small graph queries [START_REF] Lattanzi | Filtering: a method for solving graph problems in mapreduce[END_REF].

In contrast, specialized graph databases perform complex graph queries quickly, and concurrently with graph updates [START_REF]Infinitegraph: The distributed graph database[END_REF]12,13,[START_REF] Iordanov | Hypergraphdb: a generalized graph database[END_REF][START_REF] Martínez-Bazan | Dex: A highperformance graph database management system[END_REF][START_REF] Prabhakaran | Managing large graphs on multi-cores with graph awareness[END_REF]. These systems are, however, limited to a single machine (or a set of replicated images) and therefore do not scale with either the query rate, storage capacity, or data ingestion rate. Trinity [START_REF] Shao | Trinity: A distributed graph engine on a memory cloud[END_REF], a distributed graph engine, lacks support for transactional storage of graph objects. Additionally, none of these graph databases support event-driven processing. [START_REF] Fitzpatrick | Distributed caching with memcached[END_REF] stale results Distributed Slow Yes Yes, using Yes in-memory stores [START_REF]GemFire: Technical white paper[END_REF] triggers Batch graph frameworks [START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF]

Contributions

We have designed Concerto, a graph store that preserves the functionality of specialized graph databases without sacrificing the ability to scale or build eventdriven applications. As shown in Table 1, Concerto differs from existing work in its combination of two fundamental design principles.

First, Concerto provides distributed, in-memory, transactional storage of graph elements. Unlike traditional databases, Concerto embeds the graph structure within the aggregate memory of the cluster. While this layout incurs more storage than a traditional table-based layout, it enables otherwise expensive graph traversals to be performed quickly. Unlike existing graph databases, Concerto is designed to maintain data consistency across multiple servers using distributed transactions. The costs of the distributed transaction protocol are compensated by several performance optimizations: in-memory representation, fewer network roundtrips, and parallel computation.

Second, Concerto introduces the notion of graph views, which allows an application to denote subgraphs of interest. Applications can compose different views to form meaningful groups and run graph analysis on them. Applications can also register user-defined event handlers on a view. As we discuss in Section 6, some graph applications are naturally expressed as event-driven programs, and in our experience, the extensibility provided by Concerto improves performance and simplifies application programming. Moreover, views act as hints about which graph entities are related, thereby providing a means to enhance data migration, or partitioning policies [19,[START_REF] Karypis | Metis -unstructured graph partitioning and sparse matrix ordering system[END_REF][START_REF] Mondal | Managing Large Dynamic Graphs Efficiently[END_REF] and reduce communication overhead.

We empirically compare Concerto against in-memory executions of MySQL, a standard relational database; Neo4j, an open-source graph database; and Gem-Fire, a commercial in-memory distributed store. Concerto is 10× faster in bulk insertion throughput than Neo4j while consuming 3× less memory. Similarly, Concerto is more than 7× faster than MySQL for interactive k-hop query performance. Scaling results on 64 instances shows that Concerto can complete a 3-core computation of a 90-million node graph in only 12 minutes, compared to nearly 6 hours on 64 instances of GemFire. We also demonstrate Concerto's features through two real-world inspired applications: real-time incident impact analysis on a road network, and targeted advertising in a social network.

Graph storage

Concerto stores graph objects in memory and across distributed commodity servers in data centers (Figure 1a). A distributed shared memory implementation provides a global address space on which graph objects are allocated. Graph traversals take place on the distributed graph representation using server-side RPC calls batched for performance. Concerto's key contributions are in the inmemory graph representation and the use of efficient, distributed transactions to provide concurrent access and online data migration.

Graph representation

Concerto has three basic data types to store the application graph data: vertex, edge, and property. A property element contains attributes and can be attached to a vertex or edge. Vertices and edges can have multiple properties. Concerto exposes APIs to graph applications to create and update the above graph elements. New graph objects are allocated on a global address space provided by Sinfonia [START_REF] Aguilera | Sinfonia: A new paradigm for building scalable distributed systems[END_REF], a distributed shared memory system. Sinfonia exposes a flat memory region per-server called memnode which are combined to create a single global address space.

Concerto stores the logical graph using a layout optimized for in-memory reads and inserts. As shown in Figure 1b, vertices, edges, and properties are represented as records with pointers. A vertex has a pointer to a list of its outgoing edges. An edge has pointers to its source and destination vertices and to the next edge of the source vertex. Thus, all outgoing edges of a vertex can be accessed consecutively starting from the first edge. Co-locating vertices and edges in contiguous blocks of memory, and storing pointers to related graph objects allow graph traversals to be performed quickly at the cost of additional storage. Similar to edges, properties are chained together as a list. Both vertex and edge records point to the head of their property lists.

In Concerto, each vertex and edge is a fixed-size record while properties can be of variable size. Using an appropriate fixed size, a vertex or edge can be retrieved in one read transaction (one network roundtrip between a client and a Concerto server) as both the address and size of the data are known in advance. However, accessing properties of a vertex or edge may require more than one transaction. First, the vertex has to be read to determine the address of the property and then the property is read in the next transaction. In some applications, certain properties are accessed often. To retrieve these frequently accessed objects in one read transaction, properties can optionally be embedded in the vertex or edge records. Figure 1b depicts embedded properties attached to vertices and edges.

Use of transactions

Concerto uses distributed transactions to provide consistency and concurrency for graph allocation, access, and updates. Unlike simple key-value data, graph data can seldom be partitioned into shared nothing regions, and hence we need to support transactions that occur across machines. To balance consistency with efficiency, Concerto leverages a distributed compare-and-swap primitive called a mini-transaction provided by Sinfonia to support such distributed transactions. Mini-transactions are a performance-optimized implementation of the two-phase commit protocol. Concerto also provides other optimizations to minimize the number of transactions used. These include batching graph operations during traversals and reducing the number of indirections for graph object access. Using these optimizations, Concerto can, in the common case, perform reads of vertices, edges or attributes in a single network roundtrip, and finish writes in two network roundtrips. By comparison, transactionally updating even a single value in GemFire requires at least three network roundtrips. Below, we discuss examples of how transactions are used in Concerto. Graph allocation. During allocation of new graph elements (e.g., vertex, edge) it is important to ensure a unique address is assigned to the graph element even if two concurrent users request memory. Concerto uses transactions to achieve this. As shown in Figure 1c, whenever an allocation request is received, the Concerto graph allocator contacts the Sinfonia memnode. Upon allocation of an address space, an entry is made to the allocation meta-data on the memnode. Concerto wraps these operations in transactions which ensures that the metadata for the allocator remains consistent during concurrent allocation requests. Note that the use of transactions to allocate and manage each element incurs overhead, especially for vertices and edges which are only a few tens of bytes. To reduce this, Concerto pre-allocates large memory blocks from memnodes and appends new vertices and edges until the block fills up. Pre-allocated blocks reduce the amount of meta-data stored on memnodes, and also the number of network roundtrips (and possible write conflicts) from allocation requests. Figure 1d illustrates how pre-allocated blocks store vertices and edges. Graph updates. Transactions are also used to allow in-place updates to existing graph elements with other (concurrent) accesses. Internally, the Concerto transaction API calls the Sinfonia mini-transaction subsystem which allows updates to be made to graph elements on distributed machines (in this case source and destination vertices). Graph partitioning. Concerto uses transactions to provide online data migration for an application to optimize a graph partition. This can be used, for example, when adding or removing servers, or when handling data hotspots. Table 2 shows the three migrate functions available to applications. These functions implement migration as a series of tasks wrapped inside distributed transactions. For example, when migrating a vertex, the vertex and its associated data are copied to the new server, the original copy is deleted, and all incoming pointers to the vertex are updated. These tasks happen inside a transaction during which time other non-conflicting operations can continue concurrently.

Graph Views

The primary programming innovation of Concerto is the notion of applicationspecific event processing using graph views. The concept of views is well-studied in the database literature. Concerto extends this concept to distributed graph stores. In Concerto, a view is a subgraph of interest on which applications can run graph algorithms and also register generic event handlers. By using event handlers, an application is easily expressed as an event-driven program.

Programming model

Concerto provides a View class to create and manage graph views. Views are subgraphs and comprise of a list of vertices, edges, and properties. Views are primarily created to isolate regions of interest and can constrain a query to execute only on a subgraph. For example, Concerto applications use the BSP programming model to implement distributed graph algorithms by specifying code that runs on each vertex [START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF][START_REF] Valiant | A bridging model for parallel computation[END_REF]. Graph views provide application-specific

Function Description onReadVertex(V)
Invoked on read operation. Passes element where read occurred.

onReadEdge(E) onReadProperty(P) onUpdateVertex(V) Invoked on write operation. Passes element where write occurred. Old value of property also passed.

onUpdateEdge(E) onUpdateProperty(P, val)

Table 3. View API for event-driven processing.

semantics to these algorithms: applications can compose multiple views and then execute a distributed algorithm on the complex view. Consider the example mentioned in the introduction where the graph G represents a road traffic network and the graph application performs real-time accident impact analysis [START_REF] Miller | An empirical analysis of the impact of incidents on freeway traffic[END_REF] on G.

The application developer might create a view P and M corresponding to the cities of Palo Alto and Mountain View, respectively. If an accident occurs in Palo Alto, then the application can localize the execution of its impact analysis algorithm on P and demarcate the affected region I=impactAnalysis(P). Now, the application can be easily extended to provide useful functionality; to find the best path from a location in Mountain View to Palo Alto, while avoiding the impacted region, a user would simply run a shortest path algorithm on the composed view:(P -I)∪M. Concerto supports basic set operations, such as union, intersection, and subtraction, on views. For example, two views can be merged or their common elements subtracted.

Views simplify the support for event-driven processing. Applications can define a view upon which to register event handlers. The View API (Table 3) can be invoked when read or write events occur in the subgraph. For example, onreadVertex function can be invoked when a read event occurs on a vertex in the view, and onUpdateProperty function is invoked on a write event to a property element in the view. To implement function invocation, the view pointers are stored in graph elements (Figure 1b). Specifically, whenever a read or an write occurs on a graph element, the view pointer(s) associated with the graph element are traversed and the corresponding function is invoked.

Applications can invoke custom code using the View API. In our experience, read functions are broadly useful for gathering statistics and for monitoring the store. For example, the onReadVertex function can be overwritten to determine whether too many clients are reading the view members. By monitoring read throughput, data may be migrated proactively to reduce hotspots. Handlers for write events may benefit from more customization. For example, in our road traffic application, write events might specify the location of an accident, which would trigger execution of the traffic impact analysis in that region.

Data structures

Supporting graph views requires a trade-off between compact storage and fast subgraph traversal. For fast graph traversals, the ideal approach makes a copy of the subgraph corresponding to a view, enabling traversals to occur directly on the subgraph. However, this approach has serious shortcomings. For large graphs, applications may create hundreds or thousands of views. Some of these views may overlap and store (possibly) large, redundant portions of the original graph. Therefore, this approach may lead to unnecessary space explosion from duplicate copies of nodes and edges. Additionally, when updates occur, preserving the structural consistency across the views and the original graph will result in significant overhead.

To overcome these problems, the View class only stores the identifiers of its members (vertices and edges), and a scratch space to store properties about the view itself. Therefore, views store only membership information and not structural information. This storage format has the singular advantage of low space overhead. However the compact representation has the unfortunate sideeffect that by looking at only the internal representation of a view, it is not possible to traverse the subgraph. For example, the view may not contain enough information to determine the neighbors of a certain vertex. Instead, the view's stored information has to be combined with the actual graph to traverse the subgraph contained in the view.

Concerto uses hash maps to speed up traversals on a view's subgraph. Vertex and edge identifiers are hashed for fast lookups. To execute a graph algorithm on a view, the application specifies the code that runs at each vertex, but Concerto ensures that the algorithm will be constrained to only the view members.

Event-driven processing

Supporting event processing in a graph store raises several questions. Since events on a graph can span different servers, how should the graph store aggregate such information? Intercepting each event may introduce undesirable processing overhead. If so, how do we prevent event processing from unduly impacting the query throughput of the graph store? Views store the functions to be invoked when specified events occur. For example, views map the six function names in Table 3, such as onUpdateVertex, to the programmerspecified functions. Whenever events occur in a view, the runtime invokes the corresponding functions. Applications register functions to a view by calling register() on the the view. For example, V.register( onUpdateVertex, myFunc) will register the function myFunc. Concerto will invoke this function whenever any vertex is updated in the view V. Internally, the function is stored as an executable.

When the events of interest, e.g., vertex updates, occur on a graph element, the Concerto runtime needs to determine which function registered with the view should be invoked. As explained in Section 2, view members, such as vertices, have a reverse pointer to their view object. This pointer is used to reach the functions that need to be executed after the event occurs at the graph element. Figure 2 illustrates the control flow during event-driven processing. The invoked functions are first appended to a queue. These queued functions are executed by a dedicated thread pool (separate from those handling queries) and, hence, provide coarse-grain performance isolation between queries and event processing. During execution, these functions can use the Concerto runtime for read access to elements of the view. For example, after a traffic accident the impact analysis function may traverse the vertices in the view to determine the affected region. The invoked functions can also store persistent data in the property fields of the view for subsequent processing. For example, a monitoring function may store statistics about reads and writes as a property of the view.

Concerto can leverage off-the-shelf publish-subscribe systems for large-scale event propagation. We believe our work to be complementary to these systems.

Fault Tolerance and Security

Concerto simplifies the graph store architecture by delegating most of the fault recovery mechanisms to Sinfonia. Sinfonia provides atomicity, consistency, isolation, durability (ACID), and availability if replication is enabled. These guarantees are independent of client-failures and the size of the graph. This design choice ensures that the graph store can easily be ported to other platforms such as distributed key-value stores. The Concerto prototype uses Sinfonia's disklogging mechanisms to recover from memnode failures.

Sinfonia's fault tolerant global address space implies that data stored in Concerto is recoverable. However, we need mechanisms in Concerto to regain consistency (upon recovery) of the graph allocators. Graph allocators store all their meta-data in the memnodes. However, if a graph allocator fails then some of the memory may be leaked (e.g., pre-allocated blocks may be left dangling). The recovery process in Concerto goes through the allocator meta-data in each memnode and entrusts any dangling memory block to an active graph allocator.

Unlike data operations, event processing in Concerto's current prototype is not completely fault-tolerant. The difference in guarantees occurs because events are processed asynchronously to isolate query performance from event processing. As a result, an untimely fault can result in lost events. For example, when an update occurs, the write operation may return results to the client even though the triggered event processing code may still be executing a computation. If there is a fault before the update operation returns, then Concerto's recovery process will ensure that both the write operation and the event-processing code is correctly re-executed (or the client is notified of the failure and can retry). However, if a fault occurs after the update operation completes but before the event processing code completes, then the event may be lost. One can make event-processing fault tolerant by using a fault-tolerant message queue which we plan as future work.

Concerto assumes that functions registered with views are written by trusted applications. Malicious code can impact both the graph store and the stored data. The current prototype does not provide additional security features to constrain malicious functions. In the future, standard security techniques, such as sandboxes, may be used to limit the power of these functions [START_REF] Geambasu | Comet : An active distributed key-value store[END_REF]. Also, eventual consistency will let Concerto scale to more servers, improve its performance via asynchronous updates, and may decrease the latency of graph operations. However, eventual consistency is difficult to reason about and program.

Evaluation

Concerto consists of approximately 4, 500 lines of C++ code for distributed data allocation, query API, distributed graph traversals, and event processing. These lines of code do not include the Sinfonia codebase.

We compare Concerto against MySQL, a well known relational database; against Neo4j, a commercial graph database; and GemFire [START_REF]GemFire: Technical white paper[END_REF], a commercial, inmemory distributed data management platform. GemFire uses hashing to store data in memory regions distributed over multiple nodes and provides a SQL-like interface. All experiments are performed using a cluster of 100 HP SL390 servers running Ubuntu 11.04. Each server has two Intel Xeon X5650 processors (total of twelve 2.67GHz cores), 96GB of DRAM, 120GB SSD drives, and 10 Gbps NIC. In some cases, where the 96 GB memory limit was exceeded, we ran Neo4j on a separate 1 TB memory server, however 96 GB was never approached for a single Concerto memnode in these graphs.

In our experiments, all systems run in-memory: Neo4j is run on a ramfs partition, MySQL uses its memory engine, and Concerto is run without replication. GemFire is run with one logical data region, distributed over multiple nodes (the number of nodes is specified in each experiment). In MySQL and GemFire, the graph is stored as a table of edges. We optimize MySQL query performance by using B-tree and hash indexes (GemFire uses hash maps). Workload generators are located on different servers from those hosting the store.

From our evaluation, we find that:

-Concerto is fast. It can ingest millions of vertices and edges per second and is more than 25× faster than other systems for k-core on large graphs and uses 3× less memory than Neo4j. -Concerto's performance scales with the additional servers. It can calculate the 3-core on a 90 million vertex graph in less than 12 minutes on 64 instances compared to 45 minutes on a single memnode. The same computation takes more than 6 hours by GemFire. -Concerto's graph views provide extensibility. Due to views and event-processing, Concerto's implementation of a road traffic application is 10× faster than a poll based system.

Table 4 describes the graphs used in our experiments. For example, Twitter-L represents 51 million users with 2 billion follower relationships that was collected from the Twitter Web site. The Social-S/L graphs represent synthetic social network graphs generated using the model proposed by Newman [START_REF] Newman | Random graph models of social networks[END_REF].

Performance results

We first compare the insertion throughput and query latency of Concerto, Neo4j, GemFire, and MySQL.

Insertion throughput. Graph query: k-hop. A common query in many graph applications is to retrieve a vertex and its neighbors that are k-hop distance away. Figure 3 compares the latency of retrieving upto 5-hop neighbors of a vertex in the different systems.

For a 1-hop distance, MySQL and Concerto have similar performance, while Neo4j is noticeably slower. Neo4j's Java implementation is the main reason for the slowdown. However, for queries requiring more than 2-hops, MySQL is the slowest. This result is not surprising because for each hop, it has to perform a join operation, which is known to be an expensive operation for a database. For fewer Graph algorithm: k-core. The k-core of a graph determines the subgraph where each vertex has at least k neighbors on the induced subgraph. Vertices with a larger "coreness" value (i.e. k) correspond to nodes with a more central position in the network structure [START_REF] Montresor | Distributed k-core decomposition[END_REF].

The k-core of a graph is obtained by recursively deleting vertices with degree less than k, until the degrees of remaining vertices is larger than or equal to k.

For Concerto, we implement the parallel k-core decomposition algorithm [START_REF] Montresor | Distributed k-core decomposition[END_REF]. Table 4 shows the time taken by different systems to calculate the 3-core of two social graphs. For the 3 million vertex Social-S graph, MySQL, GemFire and Neo4j perform similarly, computing the 3-core of the graph in 4, 5 and 6 minutes respectively. Concerto, however, computes the 3-core much faster, requiring only 1 minute. For the 90 million vertex Social-L graph, Neo4j requires over two days to compute the 3-core, whereas MySQL and GemFire complete the same computation in 22 and 25 hours, respectively. Concerto completes the computation in only 45 minutes.

To understand these numbers, we observe that each round of the k-core decomposition consists primarily of two phases: a graph scan to find the vertices that need to be deleted, and a short traversal of each vertex to find their degree in the induced subgraph. The bottleneck in Neo4j is the part of the algorithm that must perform a scan of all the data; such scans are known to be slow in Neo4j for large graphs. MySQL's and GemFire's advantage in scans and predicate evaluation (using indexes) is the primary reason for the speedup relative to Neo4j for large datasets. We note that for the Social-L graph we used a hash index since MySQL could not create such a large B-Tree index due to a known unresolved bug in its implementation 3 . In contrast, the design of Concerto allows it to perform both graph scans and traversals quickly. It calculates the 3-core in approximately 45 minutes, which is 29× faster than MySQL, 33× faster than GemFire, and 85× faster than Neo4j.

Memory footprint

In Figure 5, we compare the storage footprint of each system, which, in this case, is entirely in memory. For MySQL we quote the numbers when only a hash index is created, which is much more memory efficient than creating a B-Tree index. Over all data-sets, MySQL has the smallest storage footprint as it stores only the edge information in the form of a table. Concerto requires 1.3 -4.7× more storage than MySQL, and requires similar storage as GemFire but is 2.8 -22.7× more space efficient compared to Neo4j. GemFire consumes less memory than Neo4j due to optimizations in object serialization and deserializationonly cached objects exist in deserialized forms, while remaining objects exist in smaller, serialized form. Apart from the overhead of Java, Neo4j also stores extra metadata and hence consumes more memory than Concerto. For example, in Neo4j the outgoing edges of a vertex are stored in a doubly linked list which incurs the additional cost of a back pointer per edge. Revisiting the performance numbers from k-hop and k-core in the context of memory footprint reveals that for simple queries (e.g., 1-hop or 3-core) over small data sets, running in-memory MySQL offers roughly the same performance-vsmemory trade-off as single-server Concerto and a better trade-off than Concerto-10. However, as the data set size or query complexity increases, the additional performance improvement of Concerto outweighs the additional storage requirement. For example, when running k-core, single-server Concerto requires 4.7× more storage than MySQL, but improves k-core latency by a factor of 29×. In the case of Neo4j, this difference is even more pronounced, where Concerto requires 3.3× less storage, and yet improves k-core latency by a factor of 85×. Compared to GemFire, Concerto consumes a similar amount of memory but is 33× faster on k-core.

View updates

Figure 6 is a microbenchmark to measure the latency in processing view updates. We created a view on the Twitter-S graph such that 20% of the vertices are part of the view (around 7M). We use a client to send randomly generated updates to vertices both within and outside the view . Whenever a vertex is updated in the view, we use event-processing to increment the count of writes occurring on the view. The Y-axis in the plot shows the time interval between a vertex update and the completion of the event handler. In Concerto-1, the latency increases substantially as the update rate becomes more than 50K/s. The increased latency is because the single server reaches full capacity utilization and incurs queuing delay. In Concerto-10, the view update latency is higher than Concerto-1 initially, due to the network communication to update the view statistic that resides at one server. However, since the graph is distributed across multiple nodes, Concerto-10 can handle a higher update rate. The average delay is under 150ms for Concerto-10 even when the update rate is 250K updates/s. As a reference point for update rates, Twitter and Facebook receive 100K-200K update events per second [START_REF]Facebook's new realtime analytics system: Hbase to process 20 billion events per day[END_REF][START_REF]Twitter by the numbers[END_REF].

Scalability results

Unlike Neo4j, Concerto can leverage distributed parallelism to improve performance. Figure 7 shows the effect of scaling on the execution time of 3-core on the Social-L graph. As we increase the number of Concerto instances to 64, the execution time drops from 45 minutes to 12 minutes. At four Concerto instances the execution time is higher than the single server case because of the extra communication required. Similarly, beyond 80 instances the communication overhead for this dataset overshadows the benefit of increasing parallelism. The table in Figure 7 shows that GemFire's performance improves with more instances. However, even at 64 instances it still requires 6 hours to complete.

Case Studies with Views

In this section, we consider how views and event-processing in Concerto can ease development and improve performance of real-world inspired applications.

Real-time traffic impact analysis

The California Performance Measurement System (PeMS) is a network of road sensors spanning the major metropolitan freeways in California; these 26,000 sensors collect data every 30 seconds, generating over 2 GB of data each day [START_REF]Caltrans performance measurement system[END_REF]. The primary challenge isn't the scale of the data, but the real-time nature of the application. We revisit the example described in sub-section 3.1 and implement an application considered by both Kwon et al., and Miller et al: a statistical technique to estimate the time and spatial impact of a road incident (e.g., an accident, obstruction) on surrounding traffic [START_REF] Miller | An empirical analysis of the impact of incidents on freeway traffic[END_REF][START_REF] Kwon | The components of congestion: delay from incidents, special events, lane closures, weather, potential ramp metering gain, and demand[END_REF]. When any such incident occurs, the application needs to react by analyzing the road network graph to predict the impact region of this incident, and possible re-calculated the shortest path between two endpoints of an impacted commute. Low latency is of the essence in order to notify the appropriate authorities to respond [START_REF] Miller | An empirical analysis of the impact of incidents on freeway traffic[END_REF].

The application leverages Concerto in three ways. First, road sensors are stored as vertices, and connecting road segments are stored as associated edges in a graph. Each vertex contains information collected by its associated sensor (i.e., traffic flow, velocity, external incidents, which are uploaded from external sources). Second, a specific region of interest -for example, a municipalityforms a graph view such that the relevant client can run analysis when events occur. Finally, a function is registered with each view to run the impact analysis algorithm upon occurrence of an incident. The analysis function can use the information contained in the sensors that span the view.

We construct a graph based on the California road network [START_REF]Stanford large network dataset collection[END_REF], and generate 10 independent views (non-overlapping subgraphs) of size 2000 to approximate independent municipalities. We drive the experiment using synthetic traffic and incident data over a 25-minute window; this data is drawn from a distribution sampled by historical PeMS data, and approximately matches the findings of Miller, et al [START_REF] Miller | An empirical analysis of the impact of incidents on freeway traffic[END_REF]. Therefore, at every 30-second interval, a traffic update invokes an updateProperty(sensorID,sensorData) in Concerto, and when an incident occurs (also associated with a specific sensor), updateProperty(sensorID, incident) is invoked. Upon completing the incident impact analysis, a shortest path between a fixed source and destination municipality is recalculated, with any impacted subgraph removed from the calculation. Figure 8 shows the end-to-end latency of determining the impact region for the first five incidents. To measure the performance benefit of using Concerto, we compare the latency of implementing this application in Concerto without eventdriven processing, thus requiring polling and client side processing (poll-based in Figure 8). We show a breakdown of the time it takes to compute the impacted region, update the incident sensor, and from queueing delay, and polling overhead. We use a 10-second polling interval, which is close to the time taken to scan all the vertices in the ten views and read whether an incident has occurred. In Figure 8 we see that Concerto can find the impacted region in less than 100 milliseconds, largely due to server-side processing in both the event handling and code execution. The polling-based system takes from 1 to 10 seconds, and thus is slower by upto two orders of magnitude from Concerto. Even discounting the polling overhead, the poll-based system takes one second to complete because of the costly client-side graph traversal.

Hotspot migration

Large social networks such as Facebook expose their infrastructure to thirdparty advertisers wishing to target particular users (e.g, through Facebook Ads API [START_REF]Facebook developers: custom audience targeting[END_REF]). An increase in targeted advertising usually coincides with increases in traffic from trending topics or external events that impact the social graph (e.g., the Super Bowl). This rapid increase in traffic can cause a workload hotspot, especially if members of the view are co-located on the same set of physical servers. Concerto can dynamically load balance data corresponding to a view to mitigate such workload hotspots. To demonstrate these features, we replay a synthetic trace of read and write traffic on the Twitter-S graph stored across three Concerto instances. To simulate peak load, we create a view (called hotspot) by randomly selecting 1, 000 Twitter users (these correspond to vertices in the Twitter-S graph) stored in the same Concerto sever. We assign a designated set of clients to this view to simulate heavy-hitters and continuously send requests to those vertices. Concerto handles the heavy hitters by using a migration policy such that the view members are evenly distributed across the three Concerto servers. This policy is implemented by gathering statistics of the number of requests hitting the view and when it exceeds a threshold (i.e hotspot occurs), the data migration policy is invoked. All of this is done using the View eventdriven processing API described in Section 3.1. Figure 9 shows the timeline for the above scenario. The hotspot occurs at time 70 seconds, at that time the average request latency seen by the heavy hitters (hotspot-avg) increases to 6ms compared to less than a millisecond initially. The average latency of other clients in the store also increases (bkgd-avg) as some of their queries are on the view data. At time 180 seconds, the migration starts and moves approximately one-third of the view members to the remaining servers. During migration 683 vertices have to be moved which requires updating 10, 916 edges. The total migration time takes approximately 19.3 seconds, representing a downtime window during which the heavy hitters cannot access the store. Note that our migration implementation isn't well optimized and the migration time of 19.3 seconds is on the higher side which can be reduced. At the end of the migration, the average latency for both the heavy hitters and the other users becomes the same. Also, as shown in Figure 10, after the migration, the store can handle more traffic from the heavy hitters as the data is now spread across more servers. The effect of this migration is reflected by the increase in the total throughput beyond time 200 seconds. Online data migration can also be used to optimize query performance in other cases. For example, in another experiment, we observe that executing 3-core on the 3M vertex graph (Social-S) is 1.6× better running on 10 Concerto servers than executing it on 32 instances. Due to the communication overhead, it takes 24 seconds to calculate 3-core on 32 instances compared to less than 15 seconds with only 10 instances. Therefore, in this case, the social media application can move the data of a view to span fewer machines before running the k-core query. Note that Concerto does not automatically partition the graph for optimal performance. But applications can use known partitioning schemes and register the partitioning logic with Concerto for dynamic data partitioning.

Related Work

Relational databases. Common graph queries such as shortest-path and k-hop are both difficult to express and inefficient to implement in a relational model. Because these queries cannot be completed quickly enough to support interactive or real-time responses. Web 2.0 sites such as Facebook, Flickr, and Wikipedia complement their SQL-based backing store with an in-memory cache such as Memcached [START_REF] Fitzpatrick | Distributed caching with memcached[END_REF] to provide low-latency response. Unfortunately the need to use a caching layer to achieve performance scalability comes at the cost of relaxed transactional semantics, a severely restricted set of supported graph queries, and lack of extensibility using application defined functions.

Distributed datastores. GemFire [START_REF]GemFire: Technical white paper[END_REF] is the most closely related system to Concerto in that it is a scalable datastore that supports parallel query processing, event-driven processing and transactions. It exports an SQL-like query interface on top of a distributed key-value storage layer. It is capable of perform-ing dynamic load balancing, event-processing over data, and in-memory caching of data objects across servers. Like relational databases and key-value stores, however, it does not provide explicit support for graph objects, thereby making graph queries inefficient.

Batch analysis. Existing approaches to large graph analysis focus on optimizing offline computation. Systems like Pregel [START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF], GraphLab [START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF], Horton [START_REF] Sarwat | Horton: Online query execution engine for large distributed graphs[END_REF] and algorithmic methodologies in the high-performance community [START_REF] Agarwal | Scalable graph exploration on multicore processors[END_REF][START_REF] Pearce | Multithreaded asynchronous graph traversal for in-memory and semi-external memory[END_REF] primarily address the challenge of scaling computation with the size of graph data, generally on the order of billions of nodes and edges. As a result, these domains restrict themselves to immutable, read-only data. On the other hand, Concerto is designed to address the challenge of providing low-latency computation with transactional semantics for complex graph problems where data is continuously ingested and modified.

Graph databases. Many specialized graph databases provide transactional guarantees and are optimized for typical graph operations, but largely do not scale storage [12,[START_REF] Iordanov | Hypergraphdb: a generalized graph database[END_REF][START_REF] Martínez-Bazan | Dex: A highperformance graph database management system[END_REF] or storage or computation [START_REF]Infinitegraph: The distributed graph database[END_REF][START_REF] Prabhakaran | Managing large graphs on multi-cores with graph awareness[END_REF] to multiple servers. Kineograph [START_REF] Cheng | Kineograph: taking the pulse of a fast-changing and connected world[END_REF] and Trinity [START_REF] Shao | Trinity: A distributed graph engine on a memory cloud[END_REF] are the most closely related graph projects to Concerto, but do not provide semantics for user-defined functions or event-based, active computation. Kineograph is designed to provide transactional support for real-time graph updates in a distributed graph storage system; however, it does not explicitly support fast graph computations and instead stores graph elements using hash-based partitioning across graph storage nodes. Trinity does not provide transactional storage, and makes a different trade-off with how edges are named and represented in the graph. In the case of InfiniteGraph, no detailed technical documentation is available to provide a more informed comparison.

Graph views. Gutierrez et. al. [START_REF] Gutiérrez | Database graph views: A practical model to manage persistent graphs[END_REF] proposed database graph views as an abstraction mechanism on relational and object oriented databases. Their work includes derivation operators such as union, intersection, difference to define new graph views. However, their work is in the context of traditional databases and they do not provide a specific implementation. Concerto's view mechanisms build upon this prior work and provide a specific implementation in a distributed environment for non-relational, in-memory graph stores.

Conclusion

Many emerging applications require both scalable, transactional data storage, and interactive, low-latency graph analysis. Concerto is a distributed graph store that fills the gap between tiered database systems that scale, but perform poorly on graph queries, and recent graph frameworks which can efficiently compute graph algorithms but are offline and don't provide transactional storage semantics. Concerto's abstraction of graph views simplifies how graph applications are expressed, and provides mechanisms that can sustain update rates reported by Twitter and Facebook.

  Internal structure of a vertex. An embedded property is optional. Example graph layout in memnode; vertices and edges are referred to by their first letters.

Fig. 1 .

 1 Fig. 1. Overview of Concerto

  its data to server s migrateView(View, s) Move view elements to server s migrateGraph(View, map) Move elements based on map

Fig. 2 .

 2 Fig. 2. Control flow of event processing.

  Fig. 3. K-hop latency. For clarity, 1 and 2 hop results are in the table.

Fig. 4 .

 4 Fig. 4. 3-core execution time.

Fig. 5 .Fig. 6 .

 56 Fig. 5. Comparison of memory usage.

Fig. 7 .Fig. 8 .

 78 Fig. 7. Distributed k-core: Concerto, GemFire execution time. (Social-L)

Fig. 9 .

 9 Fig. 9. Request latency observed during a hotspot and migration (clipped points represent downtime).

Fig. 10 .

 10 Fig. 10. Throughput observed during a hotspot and migration.

Table 1 .

 1 Comparison with competitive approaches. Concerto has multiple advantages over each system.

		Fast but offline	No	No	Yes
	Graph DBs	Fast, online	Yes	No	No
	(Neo4j [12], DEX [15])
	Trinity [17]	Fast, online	No, explicit locks	No	Yes
	Concerto	Fast, online	Yes	Yes	Yes
	Design choice →	In-memory	Distributed	Graph views,	Distributed
		data structures	transactions	notifications	processing

Table 2 .

 2 Functions to migrate data.

Table 4 .

 4 Graphs used in different experiments.

	Graph	Vertices Edges File size Experiments
	Twitter-S	33M 282M	6.5GB Insert, k-hop,
	Twitter-L	51M	2B	38GB	monitoring
	Social-S Social-L	3M 90M 405M 13M 197MB Insert, k-core 7.5GB
	Road-CA [26]	2M	5M	84MB Traffic analysis
	Inserts/sec Vertex Edge Vertex(bulk) Edge(bulk)
	Neo4j	282	337	6,120	6,467
	MySQL	21,898 15,457	504,209	324,352
	GemFire-1	5,234 6,001	153,245	165,324
	Concerto-1	6,584 7,089	1.1 million 0.9 million
	GemFire-10 27,512 23,092	1.3 million 1.0 million
	Concerto-10 29,695 27,122	2.6 million 1.8 million

Table 5 .

 5 Comparison of insertion throughput. Concerto/GemFire-1,10 represent running with 1 and 10 instances, respectively.

Table 5

 5

compares how many vertex and edge elements can be inserted per second by the different stores for the Twitter-S graph. We also compare bulk loading all the vertices and edges of the Twitter-S graph for the different stores. Bulk loading avoids overhead from multiple memory allocations by inserting vertices and edges in one request. Insertion requests are issued from multiple clients to maximize the throughput. Our results show that a single instance Concerto is more than 21× faster than Neo4j. For example, Concerto can insert more than 6,500 vertices/second compared to only 282 vertices/second for Neo4j. Inserting single data items into MySQL is faster than a single-server instance of Concerto and GemFire because MySQL is highly tuned and stores the graph in a simple table format. However, as we discuss in the next section, this table representation considerably limits graph query performance in MySQL. GemFire and Concerto exhibit similar performance with single-item insertions, but Concerto is 1.8 -7× faster than GemFire with bulk insertions. Similar to single-insertion, GemFire still needs to hash every element in the bulk insertion case resulting in lower performance. Concerto can parallelize ingestion to increase throughput. With 10 instances, Concerto inserts approximately 2.6 million vertices/second and 1.8 million edges/second. Scaling ingestion throughput is particularly useful for applications that must load very large graphs.

  3. K-hop latency. For clarity, 1 and 2 hop results are in the table.than 3-hops, the overhead from two table join operations is insignificant due to use of the MySQL index. On the other hand, graph databases are optimized for such larger traversals. For greater than 2-hops, both Neo4j and Concerto-1 are at least 2-100× faster than MySQL, with the speedup increasing with the number of hops in the query. GemFire-1 performs worse than MySQL and hence we show only the GemFire-10 numbers. While Neo4j exhibits similar or better performance than the single-server instantiation of Concerto, Concerto-10 is 2-6× faster than Neo4j and and 5× faster than GemFire-10.

	System	Social-S Social-L
	Neo4j	6 min	64 hrs
	GemFire-1	5 min	25 hrs
	MySQL	4 min	22 hrs
	Concerto-1	1 min	45 min

http://bugs.mysql.com/bug.php?id=44138

Acknowledgments. We thank the anonymous reviewers for their valuable feedback. Part of this research was sponsored by the DARPA GRAPHS program (BAA-12-01).


Patrick Reynolds

Amin Vahdat

Peer-to-Peer Keyword Search: A Retrospective

Peer-to-peer systems have been an exciting area of research. Challenges in building them have included scalability, reliability, security, and-of particular interest to these authors-search functionality. This paper surveys some of the history of the field, looks at the lasting impacts of peer-to-peer research, and provides at least one view of where we go from here.

Introduction

2001 was an exciting time for research on peer-to-peer systems. Napster [38] had recently been shut down for abetting widespread copyright violation [START_REF] Evangelista | Napster files for bankruptcy[END_REF]. Gnutella [START_REF] Ripeanu | Peer-to-peer architecture case study: Gnutella network[END_REF] and Freenet [START_REF] Clarke | A distributed decentralised information storage and retrieval system[END_REF][START_REF] Clarke | Freenet: A distributed anonymous information storage and retrieval system[END_REF] survived but used completely unstructured overlays that compromised performance and search completeness. Other peer-to-peer systems reused Napster's ideas or Gnutella's protocol, and these too were eventually shut down [START_REF] Healey | StreamCast's undoing[END_REF][START_REF] Leeds | Grokster calls it quits on sharing music files[END_REF][START_REF] Liang | The KaZaA overlay: A measurement study[END_REF][START_REF] Woody | The race to kill Kazaa[END_REF]. Peer-to-peer systems needed both better protocol design and applications other than file sharing.

Chord [START_REF] Stoica | Chord: A scalable peer-to-peer lookup service for Internet applications[END_REF], CAN [START_REF] Ratnasamy | A scalable content-addressable network[END_REF], Pastry [START_REF] Rowstron | Pastry: Scalable, decentralized object location, and routing for large-scale peer-to-peer systems[END_REF], Tapestry [START_REF] Zhao | Tapestry: An infrastructure for fault-tolerant wide-area location and routing[END_REF], and Kademlia [START_REF] Maymounkov | Kademlia: A peer-to-peer information system based on the XOR metric[END_REF] collectively introduced the idea of structured, decentralized overlay networks. The abstraction they implemented was a distributed hash table, or DHT, which mapped fixed-size, opaque keys to arbitrary values. All of the DHTs were efficient, requiring just O(lg n) operations to look up a key in an n-node system.

Soon after, distributed file systems like the Cooperative File System (CFS) [START_REF] Frank Dabek | Wide-area cooperative storage with CFS[END_REF] and PAST [START_REF] Druschel | PAST: A large-scale, persistent peer-to-peer storage utility[END_REF] were built on top of DHTs. At least initially, however, neither the DHTs nor the distributed file systems provided any search functionality.

In 2002, we set out to design a complete, efficient keyword search service [START_REF] Reynolds | Efficient peer-to-peer keyword searching[END_REF] for applications based on DHTs.

This paper revisits our original paper, surveys other interesting research on peer-to-peer keyword searching, examines the state of peer-to-peer technologies today, and identifies some of the lasting impacts that peer-to-peer networks have had.

A definition

In 2001 as well as today, the definition of a peer-to-peer system is a fuzzy one. The most prominent distinguishing characteristic of a peer-to-peer system is that nodes owned by individuals make up the bulk of both the consumers (clients) (a) A simple approach to "and" queries. Each node stores a list of document IDs corresponding to one keyword.

node sA node sB client (1) request (6) A ∩ B ∩ C (2) F (A) (5) A ∩ C ∩ F (B) node sC (3) F (B ∩ F (A)) (4) C ∩ F (B ∩ F (A))
(b) Bloom filters help reduce the bandwidth requirement of "and" queries. F (A) is a Bloom filter representing the set A, and B ∩ F (A) is the set of all elements from B that matched the Bloom filter F (A). and providers (servers) of the service. A peer-to-peer service generally uses bandwidth, storage, and/or CPU time provided by users of the service. Further, a robust peer-to-peer service should be decentralized enough that no administrator can disable the system. Some prominent peer-to-peer services, including Napster, Skype prior to 2012, and all BitTorrent search pages, rely on centralized components that can or could be administratively disabled. We still consider them peer-to-peer services, albeit ones with room for improvement.

Efficient peer-to-peer searching

A good peer-to-peer search feature needs to be decentralized, efficient, and complete. Early peer-to-peer systems were not. Napster's search feature was centralized, which made it unscalable and easy to shut down. Gnutella's search feature was both inefficient and incomplete: it flooded queries throughout the network up to a fixed number of hops away from the requester, limited by a time-to-live (TTL) value. It could not locate any resources beyond that number of hops. Early DHTs did not provide search at all.

Our contribution

The simplest implementation of keyword search on a DHT uses an inverted index, as shown in Figure 1(a). The DHT maps each keyword to a list of document IDs, corresponding to the documents that contain the keyword. A client performing a search for one keyword retrieves the list of document IDs associated with that keyword. To perform a conjunctive ("and") query of multiple keywords, the client Cache:

F (A) node sA node sB client (1) request (5) A ∩ B ∩ C (4) A ∩ C ∩ F (B) node sC (2) F (B ∩ F (A)) (3) C ∩ F (B ∩ F (A)) (a) F (A) is already in the cache at node sB. node sA node sB client request A ∩ B F (A1) F (A2) F (A3) B ∩ F (A1) B ∩ F (A2) B ∩ F (A3) A1 A2 A3 B
(b) Nodes send their data one chunk at a time until the desired intersection size is reached.

Fig. 2. Caching and incremental results

retrieves the list of document IDs for each keyword and locally calculates the intersection. This approach is clearly decentralized and complete, but it is not especially efficient. If the user searches for keywords that individually appear in many documents but that rarely appear together, then downloading the entire list of document IDs for each word is wasteful.

Our paper proposed three optimizations to this simple approach: Bloom filters, caching, and incremental results. For these to work, we changed the protocol from Figure 1(a) so that intersections are calculated within the DHT, as shown in Figure 1(b). In this revised protocol, the client sends the entire query-e.g., "efficient AND network AND protocols"-to the node hosting the first keyword. That node sends the remaining words in the query, along with a list of document IDs for the first keyword, to the node responsible for the second keyword. This second node calculates the intersection between the second keyword's set of document IDs and the set of document IDs it received from the first node. Forwarding continues in this fashion until all keywords have been considered (all sets of document IDs have been intersected), at which point the last node sends the final list-IDs of the documents containing all the keywords-back to the client.

Bloom filters [START_REF] Burton | Space/time trade-offs in hash coding with allowable errors[END_REF][START_REF] Fan | Summary cache: A scalable wide-area web cache sharing protocol[END_REF][START_REF] Mullin | Optimal semijoins for distributed database systems[END_REF] are a compact but lossy way to represent membership in a set. They answer the question "Is element x in the set," occasionally returning false positives-a value of "true" even when x is not in the set. In our search system, we used them as a compact way to represent the set of document IDs for documents that contain a given keyword. Instead of transferring entire lists of document IDs, the system transferred Bloom filters representing those lists.

Caching reduces both network traffic and latency by avoiding the transfer of information that has been used recently. Our system cached Bloom-encoded lists of document IDs corresponding to a given keyword. Figure 2(a) shows an example where node s B already has the Bloom filter F (A); the client sends the query directly to node s B , eliminating one hop and the cost of transferring F (A). Each cache hit eliminates one hop and the associated transfer cost. Caching allowed us to use larger Bloom filters, with a correspondingly lower false-positive rate.

Incremental results take advantage of the fact that users often only want a few results-say, the best ten-even when many documents match their query. At each hop, our system transferred only a few Bloom filter-encoded document IDs at a time, rather than the whole list. This optimization tied the cost of answering a query to the size of the answer the user wanted, rather than to the total number of results available. Figure 2(b) shows an example in which three chunks of the document list A are sent, and then the requested result size is reached.

In addition, our system incorporated the idea of virtual hosts [START_REF] Frank Dabek | Wide-area cooperative storage with CFS[END_REF]. Peer-topeer systems are often heterogeneous in their capabilities: nodes differ in their available CPU power and network capacity. Assigning more-capable nodes a larger number of virtual hosts allowed us to take advantage of their additional capacity.

We measured the effectiveness of our optimizations using a corpus of 105,593 HTML documents and a trace of 95,409 web searches. For each query, we calculated the number of bytes transferred and the total time for the system to satisfy the query. Taken together, our optimizations reduced the time to answer a query by about an order of magnitude.

Similar work

Other research projects tackled the problem of peer-to-peer keyword search differently. This section explores three of those systems. PIER. PIER [START_REF] Huebsch | Querying the Internet with PIER[END_REF] implements relational queries on top of a DHT, scalable to at least thousands of nodes. Relations composed of tuples are stored in the DHT; each tuple is stored according to its namespace and primary key. Joins are performed by retrieving the relevant relations with multicast queries, then storing them back in the DHT in temporary relations keyed (in the DHT) by the appropriate column for the join. Joins based on Bloom filters are provided, as well, to reduce the number of tuples that must be transferred to temporary tables.

PIER provides expressiveness well beyond keyword-index systems, and it can easily be used to implement keyword queries. However, we believe that the number of nodes involved in each query and the number of bytes transferred among those nodes will be much higher than in a purpose-built keyword-search system.

PlanetP. PlanetP [START_REF] Matias Cuenca-Acuna | PlanetP: Using gossiping to build content addressable peer-to-peer information sharing communities[END_REF] is a file sharing system built with an emphasis on searchability. Each node in the system hosts the documents its user wishes to share, as opposed to a DHT where those documents would be copied onto unrelated nodes. Each node builds an inverted index of the keywords in the documents it shares, then computes a Bloom filter to represent the set of keywords found in at least one document on that node. All nodes in the system flood their Bloom filters to all other nodes, then gossip updates as the Bloom filters change. When a user wants to perform a query, PlanetP uses the Bloom filters to figure out which remote nodes to contact. The false-positive aspect of Bloom filters adds an essentially harmless probability that some nodes will be contacted and return zero results.

OverCite. OverCite [START_REF] Stribling | OverCite: A cooperative digital research library[END_REF] is a distributed version of CiteSeer, which is a library and search engine for scientific research papers. OverCite distributes responsibility for storing, indexing, and searching papers among all participating nodes using a DHT. The responsibility for indexing those documents is divided among k partitions; the nodes responsible for a partition in the DHT index the documents contained in the partition. The rationale for using partitions is to avoid making every search query a broadcast. Each partition is 1/k the size of the full index and is replicated n/k times, so each node must store 1/k of the total index and will receive 1/n of the query load. Two additional optimizations are proposed: replicating author and title metadata to all nodes, and replicating common search terms on all nodes. Both optimizations allow certain queries to be answered by a single node, rather than sending each query to k nodes to cover all partitions.

3 Where we are now Fifteen years have passed since Napster was released, and ten years have passed since the first peer-to-peer search papers were published. Where are we now?

Very few prominent Internet services or businesses use peer-to-peer systems. The most successful peer-to-peer system by far is file sharing, accounting for 10% to 20% of traffic during peak hours, on fixed (not mobile) networks [START_REF][END_REF]. The only other peer-to-peer system that is a household name in the U.S.-clearly a subjective distinction on our part-is Spotify. Skype used to rely on a peer-topeer overlay but no longer does. Peer-to-peer networks are also popular for live and on-demand video streaming services in China.

Spotify's network is a hybrid of client-server and peer-to-peer protocols [START_REF] Kreitz | Spotify-large scale, low latency, P2P musicon-demand streaming[END_REF]. Each client keeps a persistent connection open to a Spotify server, through which it can browse available content, learn about other clients currently online, and receive the first fifteen seconds of any song where low latency is required. Desktop (not mobile) clients retrieve full songs directly from other clients whenever possible. Clients form an unstructured overlay network and use flooding queries with a TTL of two to search for songs. Spotify's central servers make the peer-to-peer protocol simpler by providing lists of online clients and by providing a backstop data source for content not present within two overlay hops. The peer-to-peer network offloads the majority of song download traffic.

Skype originally used peer-to-peer technology to provide a user directory and NAT traversal [START_REF]What are P2P communications?[END_REF]. Audio and video streams and chat messages go directly from one user to another whenever possible. Each supernode maintains a list of logged-in users and a current IP address for each one. Supernodes also assist in routing calls and chat messages to clients whose device is suspended or behind a firewall. In 2012, Skype moved all supernode functionality off of end-user PCs and onto dedicated mega-supernodes in data centers run by Microsoft [START_REF] Gillett | What does Skype's architecture do?[END_REF][START_REF] Goodin | Skype replaces P2P supernodes with Linux boxes hosted by Microsoft[END_REF].

Two large, live-video streaming services in China, PPS.tv (PPStream) and Funshion, use peer-to-peer technology. PPStream uses the DONet protocol, which is an unstructured, mesh-style multicast [START_REF] Chen | FBSA: a selfadjustable multi-source data scheduling algorithm for P2P media streaming[END_REF][START_REF] Zhang | CoolStreaming/DONet: a data-driven overlay network for peer-to-peer live media streaming[END_REF]. Funshion is based on BitTorrent [START_REF]Funshion online -about[END_REF].

Disadvantages of peer-to-peer systems

From the perspective of running an Internet service, peer-to-peer systems couple some desirable properties with some serious challenges. The most appealing property of a peer-to-peer system is that it lets a business use customers' bandwidth and computing resources without paying for them. However, in most cases, the challenges overwhelm this potential cost savings:

-Most last-mile connections have asynchronous bandwidth, heavily favoring downloads over uploads. A bandwidth-limited peer-to-peer service like file sharing or video streaming must therefore find many uploaders for each downloader. -End-user network connections, especially when geographically distant from each other, have roughly 1,000 times lower bandwidth and 1,000 times higher latency than connections within a data center. -Using customers' computers in unexpected ways can lead to bad publicity.

-Peer-to-peer services require users to download and install software, which providers must write and maintain for each target platform. An unmodified web browser can access centralized services, but it cannot access peer-to-peer services.

-Customers turn off their computers more often than data centers do.

-Mobile devices, which account for a rapidly increasing fraction of Internet traffic [START_REF][END_REF], magnify all of these issues: they are battery-and CPUconstrained, their bandwidth is usually slower and often metered, and they cannot run the same client software as desktop PCs. -Customer-owned nodes are not trustworthy.

-Services with data retention or wiretap requirements will likely find it easier to comply with the law if infrastructure is centralized [START_REF] Gallagher | Skype won't say whether it can eavesdrop on your conversations[END_REF].

Some of the limitations of customer-owned computing resources, particularly security and reliability, can be overcome with software, at a cost of additional redundancy and complexity. Others, including poor network connectivity and customers' aversion to installing additional software, are more stubborn. Overall, the risks and costs of harnessing customer-owned resources are almost always higher than the risks and costs of running services in professionally managed data centers.

Further, hosted computing, storage, and bandwidth resources have gotten dramatically cheaper per unit in the last decade [START_REF]Web hosting now vs 10 years ago[END_REF]. Utility computing, both infrastructure as a service (IaaS) and platform as a service (PaaS), allows new Internet services to start out cheaply with just a fraction of a single server, without relying on peer-to-peer systems. Both Amazon and Google offer a resourcelimited tier of hosting services for free. At this point, even cash-strapped startups favor starting new Internet services in data centers.

Web search in particular is a service that favors data centers over peerto-peer systems. Our test corpus contained 105,593 documents, while Google's contains around fifty billion. A fully featured web search service does things like spelling suggestions, autocomplete, instant search, location awareness, and personalization that are not amenable to caching or representation in Bloom filters. Our and others' search protocols mask wide-area bandwidth and latency constraints well enough for the demands of a file-sharing service, perhaps, but not well enough to be competitive with a modern search engine built with dedicated computing resources.

Advantages of peer-to-peer systems

In spite of the challenges, the most popular file-sharing services still use peerto-peer systems [START_REF]Ares[END_REF][START_REF] Cohen | Incentives build robustness in BitTorrent[END_REF][START_REF] Kulbak | The eMule protocol specification[END_REF][START_REF][END_REF]. We believe that this fact is due almost entirely to censorship resistance. No matter how widely replicated an infrastructure-based service is, it is still vulnerable to a well placed letter, phone call, domain seizure, or DMCA takedown notice. Peer-to-peer systems have proven far more resilient. In 1993, John Gilmore said, "The Net interprets censorship as damage and routes around it" [START_REF] Elmer-Dewitt | First nation in cyberspace[END_REF]. Peer-to-peer systems codify that ideal in software.

Much of the censorship exercised against web sites and peer-to-peer systems is copyright related. Publishers of movies, songs, books, and software, among others, hope to preserve their ability to charge for each copy made of their copyrighted works. However, censorship happens for other reasons, too. For example:

-Political -In 2008, both the McCain [START_REF] Fred Von Lohmann | McCain campaign feels DMCA sting[END_REF] and Obama [START_REF] Eggerton | Obama campaign spar over YouTube video[END_REF] campaigns had political messages removed due to DMCA takedown notices. Also in 2008, during the South Ossetia War, the nation of Georgia blocked all websites with addresses ending in .ru [START_REF]War still having serious impact on freedom of expression[END_REF]. -Moral -Russia has instituted an unpublished blacklist of sites relating to drugs, suicide, or pornography [START_REF]Russia internet blacklist law takes effect[END_REF]. The United Kingdom is creating its own list, currently optional but enabled by default, to restrict access to pornography; web forums; information about violence, terrorism, and eating disorders; and "esoteric material" [START_REF] Killock | Sleepwalking into censorship[END_REF].

-Security -Volkswagen successfully sued to censor research about vulnerabilities in its keyless entry systems [START_REF] Torchinsky | VW demands British court censor scientific paper about car security[END_REF]. Life science researchers have instituted policies for censoring themselves when research seems to pose more risk than social good [START_REF] Michael | Governance of dual-use research: an ethical dilemma[END_REF]. -Competitive -Universal Music Group briefly had Megaupload's "Mega Song" removed from YouTube, despite not having any copyright claim against it [START_REF] Lee | UMG claims right to block or remove YouTube videos it doesn't own[END_REF]. -Suppressing criticism -KTVU used DMCA takedown requests to remove copies of a newscast in which its anchor read obviously fake and offensive names for the pilots of Asiana flight 214 [START_REF] Kravets | Local newscast uses DMCA to erase air crash reporting blunder[END_REF]. -Accidental -The U.S. Immigration and Customs Enforcement (ICE) accidentally took down the hip-hop music blog dajaz1.com for a year before returning it without explanation [START_REF] Lee | ICE admits year-long seizure of music blog was a mistake[END_REF]. YouTube's Content ID system automatically flags videos for possible removal, but it often flags videos that do not contain copyrighted content or that might qualify as fair use [START_REF]Content ID disputes -YouTube[END_REF]. While attempting to prevent unauthorized distribution of Windows 8 and Office, Microsoft has accidentally requested that Google remove links to the BBC, Wikipedia, and OpenOffice, among others [START_REF]Microsoft DMCA notice mistakenly targets BBC[END_REF][START_REF]Microsoft censors OpenOffice download[END_REF].

In each case, censorship was possible because a small number of administrative entities-ISPs, domain registrars, YouTube, research conference organizers, etc.-could be compelled to block or remove the content. Peer-to-peer systems provide an alternative distribution channel for content when censors get too heavy handed. Formal peer-to-peer systems are not the only way that the Internet routes around censorship. Content that is small and not obviously offensive or copyrightinfringing will often end up widely distributed and widely mirrored if someone tries to censor it. Far more people know what Barbara Streisand's house looks like, how badly the Suburban Express bus company treats its customers, and how to 3D print a gun than would have if interested parties had not attempted to censor that information [START_REF]US government orders removal of Defcad 3D-gun designs[END_REF][START_REF] Gallagher | Express to Internet hate: Bus company threatens redditor with lawsuit[END_REF][START_REF] Rogers | Streisand's home becomes hit on Web[END_REF]. In a sense, technology news sites, parodies, memes, and web forums act like an informal peer-to-peer network when they mirror content in this fashion.

Impacts

Several ideas from peer-to-peer systems have found their way into systems that are not strictly peer-to-peer. Most large-scale Internet services are geographically distributed, self-organizing, and resilient. Some systems, including the Cassandra database and the Tahoe-LAFS file system, explicitly incorporate DHTs [START_REF]The Apache Cassandra Project[END_REF][START_REF]Tahoe-LAFS[END_REF].

BOINC-the computing platform that runs SETI@home [6]-runs primarily on end users' computers, much like a peer-to-peer system. Unlike nodes in a peer-to-peer system, nodes in BOINC are only servers and do not consume the service that other nodes provide. Also, nodes are centrally managed rather than self-organizing. However, like a peer-to-peer system, BOINC successfully deals with malicious participants and harnesses spare CPU cycles. [START_REF]US government orders removal of Defcad 3D-gun designs[END_REF] Where we go from here

We believe that peer-to-peer systems continue to have both technical and social value. They may be a good way for modestly funded research groups to bootstrap an Internet service. They provide a stress test for new protocols, because protocols and techniques that work within the resource and security constraints of a peer-to-peer system will often work even better in a centralized system. Finally, of course, they provide outstanding resistance to censorship, in a way that commercial and centrally managed services cannot.

To that end, we believe the most important focus areas in peer-to-peer research are:

-Security -Services that become popular, or that host content that someone wishes to censor, become the target of attacks. Attackers may disrupt routing or searching, or they may intercept or modify content. Peer-to-peer systems have to deal with the possibility that nodes are Byzantine faulty [START_REF] Lamport | The Byzantine generals problem[END_REF]. -Anonymity -If the main use case for peer-to-peer systems is distributing censored content, then participants might need to remain anonymous when publishing or retrieving that content. Providing anonymity in a robust, efficient way could provide immense social value. -Searchability -Simply put, we cannot read what we cannot find. BitTorrent users currently rely on centralized, commercial web sites to map keywords to document identifiers. Eventually, whether through legal changes or technical attacks, these sites will probably get shut down. Existing and new peerto-peer search technologies should be applied to ensure that users can find content.

In 2013, as in 2001, peer-to-peer networking remains an exciting, fruitful area of research.

Fig. 1 .

 1 Fig. 1. Adding a Bloom filter to DHT search


S Erard

Z Yin

B Cecconi

Lesia

INTEROPERABLE MARS ATMOSPHERE DATA SERVICES

Introduction:

Modern space borne instruments produce huge datasets, especially on long-lived missions such as Mars-Express. This calls for new ways to handle the data, not only to perform mass processing, but also more basically to access them easily and efficiently. Virtual Observatory (VO) techniques developed in Astronomy during the past 15 years can be adapted to address this problem, provided they are enlarged to include specificities of Solar System studies such as coordinate systems, target-related time scales (local time and season), or measured quantities. An effort to adapt VO techniques to Solar System studies has been started in the frame of the Europlanet program, first in FP7 (2009-2012) as a demonstrator, and now in Horizon2020 (2015-2019) in a more extensive way. In the current program Europlanet 2020, the VESPA activity deals with the infrastructure and implements new data services. VESPA stands for Virtual European Solar and Planetary Access and supports all aspects of Solar System science. Its developments are essentially based on the standards of the IVOA (International Virtual Observatory Alliance).

As part of this enlargement, several data services related to the Martian atmosphere were recently designed and released, including a set of SPICAM derived data and a demonstrator of VO access to the Mars Climate Database. In both cases, vertical profiles are distributed with the basic VESPA architecture. In this abstract, we present the two services and assess this type of access to a simulation service.

Architecture:

The VESPA data access architecture [START_REF] Erard | VESPA, a Virtual Observatory infrastructure for Planetary Science and Solar System studies[END_REF] is based on a new data access protocol, a specific user interface to query the available services, and intensive usage of tools and standards developed for the Astronomy VO [START_REF] Erard | Planetary Science Virtual Observatory architecture[END_REF]. The Europlanet data access protocol, EPN-TAP, relies on the general TAP (Table Access Protocol) mechanism associated to a set of parameters that describe the contents of a data service [1] [3]. These parameters are defined to enable queries on quantities relevant to the scientific user, including observational and instrumental conditions, and are defined to handle the specific diversity and complexity of Planetary Science. Services are required to return answers formatted as VOtables, which are handled by all standard VO tools.

Data services are installed at their respective pro-viders institutes and are declared in the standard IVOA registries, so that they are always visible and reachable from query interfaces. Although accessible in many ways, EPN-TAP data services are best queried from the optimized VESPA user interface: http://vespa.obspm.fr

In the frame of TAP, all data services present a list of granules (usually data files) described by a series of parameters. The Europlanet data access protocol, EPN-TAP, defines a set of mandatory parameters introducing metadata that describe all granules; this is similar to the ObsTAP protocol from IVOA, which describes observational datasets in Astronomy. EPN-TAP metadata introduce both observational and instrumental conditions: ranges on several axes (spatial, temporal, spectral, photometric), measurement type, origin of data, and various references. Location is provided in the most appropriate coordinate system (e.g., sky or planetary coordinates); target-related time (local time and season, through Ls) can be provided when relevant. The VESPA interface uses the mandatory parameters to search for individual granules in all registered data services at once, allowing for discovery of data content unknown to the user. In addition, specific parameters may also be used to describe services in more details, and can be used to identify granules more precisely when querying a single service.

All granules provide a link to a data file, or include the data itself in the table when possible (e.g., for scalar quantities). Data description parameters are used to identify adequate VO tools to access, plot and handle the data correctly. They not only provide a description of the file format, but also specify the dimensions, units, and physical quantities, relying on IVOA data models extended for VESPA. For instance, spectra and images are handled in different tools, and spectra measured in radiance or in reflectance are handled differently by the spectral tools.

Mars data services content:

Currently, two such data services are implemented for the Mars atmosphere. Profiles measured by SPICAM during the first Martian year of Mars-Express operations [START_REF] Forget | Density and temperatures of the upper Martian atmosphere measured by stellar occultations with Mars-Express SPICAM[END_REF] [5] are distributed through EPN-TAP in a standard way. The SPICAM service contains 616 temperature & CO 2 profiles plus 433 O 3 profiles, with arbitrary spatial distribution. It will be completed with aerosols profiles in the next future [START_REF] Määttänen | A complete climatology of the aerosol vertical distribution on Mars from MEx/ SPICAM UV solar occultations[END_REF]. Simulated vertical profiles from the Mars Climate Database (MCD) [START_REF] Navarro | Global climate modeling of the Martian water cycle with improved microphysics and radiatively active water ice clouds[END_REF] [8] are accessible in two different ways. First, as a standard EPN-TAP service providing situations sampled on a regular grid of coordinates, local time and season, for some of the available scenarios (a total of 62,244 profiles in this demonstration step). While the MCD supports quasicontinuous variations in input parameters, sampling is required to implement data access via the TAP mechanism, which is originally designed to provide access to catalogues. Although the resolution of the service cannot be maintained in these conditions, a trade-off has been searched between the physical size of the resulting data service and the accuracy of the results. A second VO access to the MCD is discussed below.

The profiles themselves are provided as VOtables, a handy format for this type of data that is directly recognized by VO tools. Parameters included in MCD profiles are temperature, pressure, densities of H 2 O, O 3 , and CO 2 computed in 50 layers up to 250 km altitude, plus column densities of water vapor and water ice, and dust opacity. SPICAM profiles include 3 different sets of temperature and CO 2 profiles in the 60-140 km range, derived under 3 different assumptions on the temperature at the top of the atmosphere [START_REF] Forget | Density and temperatures of the upper Martian atmosphere measured by stellar occultations with Mars-Express SPICAM[END_REF]. SPICAM measurements of O 3 are available separately in the 20-70 km range [START_REF] Lebonnois | Vertical distribution of ozone on Mars as measured by SPICAM/Mars-Express using stellar occultations[END_REF]. The vertical scale is provided as radial distance and as altitude above both the areoid and the surface (i.e., including MOLA topography) for the two services.

Search interface and basic tools:

The first function of the VESPA interface is of course to provide search capacities in the datasets; in this case, the obvious search parameters are location, season, local time, plus the scenario adopted for simulations. Since all VESPA services are queried simultaneously, this readily identifies corresponding situations in the observed and simulated datasets.

The distribution of answers to such queries can be visualized rapidly by plotting the metadata. Those can be sent from a service result page, either for all the answers or for a manual selection in this list; they are smoothly transferred from VESPA to VO tools according to the IVOA SAMP protocol, with no user intervention. MIZAR and Aladin immediately display the footprint of SPICAM profiles on a 3D view of Mars; TOPCAT can overplot other metadata such as local time or season on a 3D sphere.

Fig 2: Footprints of selected profiles (punctual in this case) are directly plotted by Mizar in a browser

The data themselves can be transferred in a similar way. This allows the user to overplot observations and simulations with minimum manipulations in TOPCAT, the displaying tool of choice for profiles.

Fig 3: All footprints of SPICAM O 3 profiles plotted together with the current sampling grid of MCD in TOPCAT

Comparisons:

As seen on Fig. 3, the observations are actually grouped in a small part of the surface, which is only scarcely sampled in the current MCD service. The observations are even more closely grouped on the other two parameters, Ls and local time, so that finding a simulation really comparable to the measure-ment is difficult with the current sampling steps. For this reason, we're using here profiles extracted from the MCD with exact parameters, through the other mechanism described later on.

We're mimicking here a study of SPICAM and the MCD published in [START_REF] Forget | Density and temperatures of the upper Martian atmosphere measured by stellar occultations with Mars-Express SPICAM[END_REF] and we're interested in comparing the two plots, regardless of the goodness of fit (Fig. 4). Local variations have been smoothed out in [START_REF] Forget | Density and temperatures of the upper Martian atmosphere measured by stellar occultations with Mars-Express SPICAM[END_REF] by computing monthly averages in extended areas, which yields a reasonable match. Conversely, we're comparing single profiles as extracted from the VO services using the averages of the parameter rangesthis is a matter of minutes. Although this comparison validates the content of the services and query system, we notice that our fits are significantly poorer and probably less informative. This stresses the need to develop a procedure to average the retrieved profiles, which is currently difficult in TOPCAT.

Fig 4: SPICAM profiles compared with MCD simulations:

the large plot uses individual profiles retrieved from the VO services, the inlay from [START_REF] Forget | Density and temperatures of the upper Martian atmosphere measured by stellar occultations with Mars-Express SPICAM[END_REF] uses averages.

Discussion:

Beyond the usual added value of VO access to observational data, this use case demonstrates the potential of comparing observations and simulations entirely in a VO environment. However, access to complex simulations via a strict TAP procedure appears to be restrictive; this requires sampling the parameters space and precomputing files corresponding to these situations. If the simulation uses many parameters, this is rapidly becoming inadequate: storing the files requires a large disk space but most of all, the details of the simulation are lost in the sampling procedure. This is clear in the current example: although the MCD sampling step is much too coarse to preserve the accuracy of the simulation (Fig. 5), it already results in a significant data volume (~ 100 GB of precomputed files to accommodate all scenarios at the current resolution). Although a moderately sampled EPN-TAP service such as the current one is valuable for public outreach and teaching purposes, increasing the resolution of the service is required to meet the needs of scientific users, say by a factor of 10 on each parameter, but this would lead to impracticable data volumes.

For this reason, a second VO-like access to the MCD has been implemented. This service launches a script that triggers on-line computation with the parameters provided in the query, similarly to the standard on-line interface of the MCD. The same script formats the resulting profile as a VOtable, which is directly ingestible in TOPCAT and other VO tools. TOPCAT provides an acceptable interface to this service, since it can call the script on the server and download the result. The current EPN-TAP service also makes use of this script system, which eliminates the need to store precomputed files and therefore allows for much smaller sampling steps (at the expense of more computation on the server side). Although the script can be called "manually" from VO tools, the EPN-TAP layer takes advantage of the querying syntax of the VO and TAP: it allows the user to search for various configurations of input parameters, and presents the results in a table identical to that of observational data. However, the available resolution of the EPN-TAP service is still somewhat limited by the table size (currently 500 MB, and could amount to some TB to be handled in postgreSQL). In order to preserve the flexibility of TAP access while maintaining the full resolution of the simulation, a different calling system is being studied. The next step will be to convert the EPN-TAP query issued by the VESPA interface into a direct call to the computing service. This would accept continuous values of the input parameters as the usual MCD interface does, or it could perform averages if ranges of parameters are provided. VESPA will experiment with scientific workflow engines already used by the IVOA, and study a possible implementation of this scheme. Other MCD outputs will be distributed in a similar way, e.g. local diurnal temperature curves.

APPENDIX -Other atmospheric services in VESPA

VESPA currently connects 27 data services in all areas of Planetary Science, Heliophysics, and Exoplanets, and about as much are being designed. In the area of planetary atmospheres, open services include SOIR profiles and VIRTIS calibrated data on Venus, CIRS/Cassini profiles on Titan, and absorption cross-sections of various molecular species in the UV-visible range. Besides, a workflow engine will be implemented to allow for processing on demand. In addition to the MCD case described here, possible usages for planetary atmospheres include radiative transfer codes or inversion procedures applied to entire data services.

APPENDIX -VESPA prospects and development plan

VESPA compliant data services are expected to multiply in the coming years, to include data archives on one hand, and derived data on the other hand. Agreements or projects exist with ESA to make the whole PSA content available in this system; with PDS small bodies node to implement a bridge toward (recent) PDS4 archives, which could be extended to the other PDS nodes; with other space agencies which are also considering VESPA as an alternative way to make their data available. This will provide the science users with a handy way to search large datasets and easily identify data of interest, based on observational or operational conditions.

Distributing derived data is of course even more profitable to the community. Although this can be coordinated at the level of a whole mission, typically by the IDS (e.g., on Mars-Express or Cassini), research teams or individual researchers can also contribute and take advantage of the system directly by setting up a small data service to distribute results of analyses, e.g., from a published paper. All the VESPA documentation is freely available (although still partly in progress at this point). But most of all, VESPA opens a yearly call for new services; selected teams are invited to a 4-days implementation workshop where they will acquire the know-how, install their own data server, and design their service.

These calls are advertised in particular on the VESPA web site: http://www.europlanet-vespa.eu

Fig 1 :

 1 Fig 1: Result page for a query on SPICAM service

Fig 5 :

 5 Fig 5: SPICAM profiles compared with MCD simulations: with exact parameters, or from closest sample. The sampled configuration is much further from the observations (Ls=13.2, LocalTime= 21.7, lon=247.8, lat= 44.4)

Acknowledgements:

The Europlanet 2020 Research Infrastructure project has received funding from the European Union's Horizon 2020 research and innovation programme under grant agreement No 654208.

Additional funding was provided in France by the Action Spécifique Observatoire Virtuel and Programme National de Planétologie / INSU.