Methods and systems for using dispersed cached data stored in multiple database nodes for serving database access requests are described herein. Upon receiving a request for data from a requesting device, a first application server determines whether the requested data is stored in a local cache memory. If it is determined that the requested data is not stored in the local cache memory, without accessing a local, first database, the first application server determines that the requested data is stored in a cache memory of a second application server, wherein the second application server stores at least a portion of the data from a second database in its cache memory. The first application server retrieves the requested data from the cache memory of the second application server and provides the retrieved data to the requesting device.

BACKGROUND

The present specification generally relates to database systems, and more specifically to using dispersed cached data to improve database performance.

RELATED ART

As online activities and transactions become more prevalent, demands for constant and uninterrupted online services are increasing. As a result, enterprise database systems are expected to perform at a higher level than ever. For example, many enterprise database systems today are expected to serve hundreds of millions of requests each day. The reliance on constant and uninterrupted online services also means that unavailability of a database system, no matter how short the downtime, can be costly to a business. As an example, several seconds of downtime for a database system may translate to denying or failing to process thousands of requests from end users. Thus, enterprise database system developers are facing pressure to improve the availability performance of database systems.

Data caching has been widely used to improve the speed for serving database access requests. However, most caches are not able to refresh when the corresponding database is unavailable, and due to the limited amount of high-performance memory, a large portion of data from the database is not cached. Therefore, the benefits of data caching is not fully realized. Thus, there is a need for improved database systems and methods that provide improved availability and speed performance.

DETAILED DESCRIPTION

The present disclosure describes methods and systems for serving database access requests for a database system using dispersed cached data that is stored on different nodes associated with the database system. In some embodiments, the database system may include multiple nodes communicatively coupled with each other over a network. The nodes may be implemented within computer servers that are located in different geographical areas (e.g., in different cities across a country, in different countries, etc.) configured to serve database access requests from the corresponding geographical areas. Each node may include, or may have access to, a database storing data for the database system. In some embodiments, the databases at the nodes store a common set of data.

Each node in the database system may also include an application server that is communicatively coupled to the corresponding database of the node. In some embodiments, the application server interacts with external devices (e.g., end-users) and the database to serve database access requests. Each of the application servers may also include a cache memory for providing fast data retrieval such that the database access requests may be served more efficiently. For example, the application server may store a copy of the data that has just been requested in the cache memory such that in subsequent requests of the same data, the application server may retrieve the data from the cache memory instead of retrieving from the corresponding database. According to various embodiments of the disclosure, each application server may use data stored in its local cache memory and also data stored in cache memories of other application servers to serve database access requests. This way, the collective data stored in the cache memories of the application servers within the database system may act as a micro-database such that at least some of the database access requests may be served more efficiently and even when one or more databases are unavailable. When the number of nodes, and their corresponding application servers, is sufficiently large, the collective data may represent a large portion (or even the entirety) of the dataset stored in the databases. As a result, many or all of the database access requests may be served without accessing the databases themselves.

In some embodiments, the application server is a separate component from the corresponding database, such that the application server may still respond to database access requests from external devices when the corresponding database is unavailable (e.g., the connection with the database fails, the database is offline, etc.). According to various embodiments of the disclosure, upon receiving a request for data from a requesting device, the application server may initial determine whether the requested data is stored in the local cache memory of the application server. If it is determined that the requested data is stored in the local cache memory of the application server, the application server may retrieve the requested data from the local cache memory and may transmit the requested data to the requesting device. However, if it is determined that the requested data is not stored in the local cache memory, the application server may determine that the requested data is stored in a remote cache memory of another application server (e.g., of another node). The application server may then retrieve the requested data from the remote cache memory of the other application server via a network, and may transmit the requested data to the requesting device. Thus, the application according to various embodiments of the disclosure is capable of serving database access requests even when the corresponding database is unavailable, thereby improving performance of existing database systems.

FIG. 1illustrates a database system104according to various embodiments of the disclosure. As shown, a database system104is communicatively coupled with external devices, such as a requesting device102via a network160. The network160, in one embodiment, may be implemented as a single network or a combination of multiple networks. For example, in various embodiments, the network160may include the Internet and/or one or more intranets, landline networks, wireless networks, and/or other appropriate types of communication networks. In another example, the network160may comprise a wireless telecommunications network (e.g., cellular phone network) adapted to communicate with other communication networks, such as the Internet.

The database system104may include multiple nodes, such as a node106and a node108. While only two nodes (nodes106and108) are shown in this figure, it has been contemplated that the database system104of some embodiments may include as many (or as few) nodes as needed. As discussed above, the multiple nodes may be implemented within different physical machines (e.g., a computer server) that are located in different areas. In some embodiments, the different nodes may be located in different cities within a country, or in different countries, for serving database access requests initiated by devices located in the corresponding regions. In particular, each node may be configured to serve database access requests initiated from a predetermined geographical region (e.g., a city, a state, a country, etc.). Each node is also communicatively coupled with each other, and with other external devices, such as the requesting device102, via the network160. The requesting device102may be any type of computing device, such as a personal computer, a smart phone, or a tablet, being operated by a user. Although only one requesting device is shown in the figure, it has been contemplated that more requesting devices from different geographical regions may be used by end-users to transmit database access requests to the database system104.

Through the requesting device102, a user may perform an online activity (e.g., browsing the web, searching for an item, making an electronic transaction, etc.) over the network160. The online activity may include transmitting a database access request to the database system104(e.g., accessing an item offered by an online merchant, etc.). Based on one or more factors (e.g., the geographical location of the requesting device102), the database access request may be routed to one of the nodes106and108for servicing the request.

In some embodiments, each node includes a database and an application server. For example, the node106includes a database116and an application server110, and the node108includes a database126and an application server120. In some embodiments, each database is locally connected with the associated application server within the node. For example, the database116may be housed within the same physical server as the application server110or connected with a physical machine associated with the application server110within an internal network (e.g., local area network), and without going through an external network such as the Internet. In other embodiments, multiple nodes may access the same database. For example, the node108may not include the database126, but instead, may access the database116. In these embodiments, the database116may be located within the node106(locally connected to the application server110) or may be located externally from both the nodes106and108but accessible by both of the application servers110and120. It is noted that while the description below refers to an example where each node includes its own database, it has been contemplated that the techniques described herein may be operated in an environment where at least some of the application servers within the database system104access the same database.

Each of the databases116and126may store and maintain various types of information for use by the corresponding application servers110and120, and may comprise or be implemented in one or more of various types of computer storage devices (e.g., servers, memory) and/or database structures (e.g., relational, object-oriented, hierarchical, dimensional, network) in accordance with the described embodiments. It is noted that each of the databases116and126may store data in a persistent/non-volatile data storage (e.g., a flash drive, a hard drive, etc.) such that the data stored in the databases116and126may persist over power cycles of the databases116and126. In addition, each of the databases116and126may include a query engine configured to produce a result dataset based on a query (e.g., SQL query, etc.). In some embodiments, at least a portion of the data stored in each of the databases116and126are common with each other. For example, a dataset corresponding to the database system104may be replicated across the databases116and126. Whenever there is an update to the data stored in one database, the corresponding data in the other database(s) may also be updated, for example, by implementing a periodic synchronization process across the databases. Furthermore, the databases116and126may be different types of databases. For example the database116may include a relational database management system (RDMS) while the database126may include an object-oriented database.

Each of the application servers110and120may be implemented as one or more stand-alone server machines. Exemplary servers may include, for example, enterprise-class servers operating a server operating system (OS) such as a MICROSOFT® OS, a UNIX® OS, a LINUX® OS, or another suitable server-based OS. It can be appreciated that the application servers illustrated inFIG. 1may be deployed in other ways and that the operations performed and/or the services provided by such servers may be combined or separated for a given implementation and may be performed by a greater number or fewer number of servers.

Each of the application servers110and120includes a database management module and a cache memory. For example, the application server110includes a database management module112and a cache memory114, and the application server120includes a database management module122and a cache memory124. In some embodiments, each of the database management modules112and122may serve incoming database access requests (e.g., a read request or a write request) initiated by one or more requesting devices. Based on an incoming database access request, each of the database management modules112and122may generate a query, may run the query against the corresponding database (database116or database126) to obtain a result dataset, and may transmit the result dataset back to the requesting device.

In addition, each of the database management modules112and122may cache at least some of the data from the corresponding database (database116or database126) in the corresponding cache memory (the cache memory114or the cache memory124). Each of the cache memories114and124includes temporary storage space or memory (persistent or non-persistent memory) (e.g., dynamic random-access memory (DRAM)) that provides fast access to data. It can be implemented by hardware or a combination of software and hardware. Typically, each of the cache memories114and124has smaller storage capacity than its corresponding database (the databases116and126). As such, each of the database management modules112and122may only store a portion of the data from the corresponding database in the corresponding cache memory for fast retrieval. The database management modules112and122may use one or more cache policies for determining which data to store in the cache memories114and124. Example cache policies may include a policy of caching most frequently accessed data, a policy of caching most recently accessed data, and the like, or a combination of any of the policies. It is noted that each of the database management modules112and122may implement its own cache policies independent of each other. As such, each of the application servers110and120may cache data according to its needs. For example, the application server110may cache data that is most frequently accessed at the application server110, while the application server120may cache data that is most recently accessed at the application server120. In other words, the application servers110and120do not coordinate with each other in caching data, which is different from typical implementations of distributed cache systems. For example, as will be discussed in more details below, each application server may store data in its cache memory according to its own cache policy, without sharing information about the cache policy with the other application servers in the database system104. The cache policies for different application servers may even be different from each other. As such, each application server may store data in its own cache memory independent of what data may be stored in the cache memories of the other application servers. Without the coordination in caching, the database system104according to various embodiments of the disclosure advantageously removes the extra layer of complexity required by a typical distributed caching system.

Even though the caching of data at the different application servers is uncoordinated, it has been contemplated that different data may be cached at different application servers. In other words, the collective data that is cached at the different application servers is larger than the data that is cached at one application server. Given a sufficiently large number of nodes (and application servers) (e.g., 10, 50, 100, etc.), the collective data that is cached at the different application servers may comprise a large portion (or close to the entirety) of the dataset stored in each of the databases. It has been contemplated that the collective data that is cached at the different application servers may be used by the application servers as a micro-database for serving database access requests such that access to the databases may be reduced. In some embodiments, using this approach, database access requests may be served even when one or more local databases are unavailable.

According to various embodiments of the disclosure, when an application server (e.g., the application server110) receives a database access request (e.g., a read request for a data from the database116) from a requesting device (e.g., the requesting device102), the application server may first determine whether the requested data is stored in a local cache memory (e.g., the cache memory114). If the requested data is stored in a local cache memory, the application server may retrieve the requested data from the local cache memory and may transmit the requested data to the requesting device. If the requested data is not stored in the local cache memory, instead of querying the corresponding database (e.g., the database116), the application server may determine which other application server (e.g., the application server120) stores the requested data in its own cache memory. For example, the application server110may determine that the requested data is stored in the cache memory124of the application server120. The application server110may then retrieve the requested data from the cache memory124(e.g., by sending a request for the requested data to the application server120) via the network160, and transmit the retrieved data to the requesting device.

The application server may use different mechanisms to determine which other application server stores the requested data in its cache memory. In one embodiment, the database system104may maintain a cache index, such as a cache index130that stores information indicating which application server(s) stores a particular data in its cache memory. Different embodiments may implement the cache index130in different ways. For example, the cache index130may be implemented as a hash table where dataset keys may be hashed to generate a value that corresponds to an identifier of an application server. The cache index may be implemented within a standalone server (either within any one of the nodes106and108or in a separate server) or may be implemented in a distributed manner across the application servers110and120, for example, using a distributed data configuration management such as Zookeeper®. In some embodiments, each application server is responsible to update the portion of the cache index corresponding to data stored in its local cache memory.

FIG. 2illustrates servicing a database access request by the database system104according to various embodiments of the disclosure. In this example, a request for data having a key ‘ABC’ is initiated by the requesting device102and transmitted to the application server110. As discussed above, each of the application servers within the database system104may be configured to service requests from a corresponding geographical region. As such, the request for data may be transmitted to (or routed to by the database system104) the application server110, among all of the application servers of the database system104, based on the geographical location of the requesting device102. The request for data may also come in different forms. For example, the request for data may come in the form of a universal resource locator (URL) request. In such an example, the user of the requesting device102may type in the URL or selecting a link of an existing page (e.g., a webpage of the database system104).

Upon receiving the request for data, the application server110(for example, the database management module112of the application server110) may determine whether data value (the requested data) having a key ‘ABC’ is stored in the local cache memory (e.g., the cache memory114). When it is determined that the requested data is not stored in the cache memory114, the application server110may determine which (if any) other application server stores the requested data in its cache memory. In some embodiments, the application server110may make such a determination by looking up the cache index130based on the key ‘ABC.’

FIG. 2illustrates an example implementation of the cache index130. As shown, the cache index130includes multiple entries of key and network address pairs. Each key is mapped to a network address corresponding to an application server of the database system104that has the data value corresponding to the key stored in its cache memory. In this example, the cache index130includes a first entry having a key ‘ABC’ with a corresponding network address of ‘168.45.233.154,’ which corresponds to the application server120. The cache index130also includes a second entry having a key ‘GHI’ with a corresponding network address of ‘268.432.543.326,’ which may correspond to the application server110. While only two entries are shown, more entries (or any number of entries) may be stored in the cache index130as needed.

As such, based on the key ‘ABC,’ the cache index130may return a network address (e.g., IP address 168.45.233.154) corresponding to an application server (e.g., the application server120) that stores the requested data in its cache memory. Using the network address received from the cache index130, the application server110may transmit a request for the data to the application server120. In response to the request sent from the application server110, the application server120may retrieve the requested data from the cache memory124and may transmit the requested data (e.g., the data value corresponding to the key ‘ABC’) to the application server110. Upon receiving the requested data from the application server120, the application server110may transmit the requested data to the requesting device102.

In the example discussed above, the cache index130indicates that the data value corresponding to the key ‘ABC’ is stored only in the cache memory of one application server within the database system104. Since the caching of the application servers are performed in an uncoordinated way, it has been contemplated that in some instances, the same piece of data (e.g., the data value corresponding to the key ‘ABC’) may be stored in cache memories of multiple application servers within the database system104. For example, the cache index130may indicate that the requested data is stored in the cache memories of the application server120and a third application server that store the requested data in their cache memories. In this scenario, different embodiments of the application server110may use different approach to select one application server to use for retrieving the requested data. Under one approach, the application server110may determine a remote application server having the fastest response time, and retrieve the requested data from such a remote application server. The application server110may make such a determination by sending a request signal (e.g., a ping) to each of the application servers having the requested data in their cache memories, where the request signal requires the corresponding application server to respond with a response signal. The application server110may then determine which application server has a faster response time by comparing the speed in which the remote application servers respond to the request signals (e.g., the network speed). The network speed may be affected by the bandwidth between two application servers, the traffic between two application servers, or other network factors. Under another approach, the application server110may retrieve the requested data from a remote application server that has the most updated requested value. In this scenario, the index cache130may also include a time when the data was stored in the cache memory of the respective application server, in addition to the key and the network address. Thus, the application server110may determine a remote application server that stores the requested data in its cache memory most recently based on the time in the entries, and may then retrieve the requested data from such a remote application server.

FIG. 3illustrates a process300for using dispersed cached data stored in different application servers to service a request for data according to various embodiments of the disclosure. The process300may be performed by an application server (e.g., the application server110of the database system140). The process300begins by receiving (at step305) a request for data from a requesting device. For example, the request for data may be initiated by a requesting device (e.g., the requesting device102) and transmitted to the database system104. The request for data may be transmitted directly to an application server of the database system104(e.g., the application server110) or may be routed to the application server based on one or more criteria (e.g., geographical location of the requesting device). As the application server110receives the request for data, the application server110may optionally determine (at step310) that the local database is unavailable. The application server110may determine whether the database116is available or not by sending a request signal (e.g., a ping, a data access request, etc.) to the database116and detect a response signal from the database116. If the application server110receives a normal or expected response from the database116in response to the request signal, the application server110may determine that the database116is available. However, if the request signal times out (no response after a predetermined duration) or if an error signal is received, the application server110may determine that the database116is unavailable. In some embodiments, the process300performs the following steps (e.g., steps315-335) in response to the determination that the local database is unavailable. In these embodiments, when the local database is available, the process300may retrieve the requested data from the local database and may provide the requested data to the requesting device.

However, while the process300may be performed to advantageously service requests for data when a local database is unavailable as discussed above, it has been contemplated that the process300may be performed even when the local database is available such that the requested data may be retrieved from a local cache memory or a remote cache memory to efficiently service the requests. Thus, in some embodiments, the process300may not perform the determining step at step310. The process300then determines (at step315) whether the requested data is stored in a local cache memory (e.g., the cache memory114of the application server110). If it is determined that the requested data is stored in the local cache memory, the process300retrieves (at step320) the data value corresponding to the request for data from the local cache memory and provides (at step335) the requested data to the requesting device.

On the other hand, in response to a determination that the requested data is not stored in the local cache memory, the process300determines (at step325) a remote application server that has the requested data in its cache memory. For example, the application server110may use a key from the request for data to look up a network address corresponding to a remote application server from the cache index130in a manner discussed above by reference toFIG. 2. In the event that more than one remote application servers has the requested data stored in their cache memories, the application server110may use different ways to select one remote application for retrieving the requested data as discussed above by reference toFIG. 2. Once a network address of the remote application server is retrieved, the application server110may retrieve (at step330) the data value corresponding to the request for data from the cache memory of the remote server (e.g., from the cache memory124of the application server120), and may then provide (at step335) the retrieved data to the requesting device.

FIG. 4illustrates a process400for updating a cache index according to various embodiments of the disclosure. The process400may be performed by an application server (e.g., the application server110of the database system140). The process400begins by receiving (at step405) a request for data from a requesting device. For example, the request for data may be initiated by a requesting device (e.g., the requesting device102) and transmitted to the database system104. The request for data may be transmitted directly to an application server of the database system104(e.g., the application server110) or may be routed to the application server based on one or more criteria (e.g., geographical location of the requesting device). Upon receiving the request for data, the application server110determines (at step410) that the local database is available. In some embodiments, if the application server110determines that the local database is unavailable, the application server110may switch to perform the process300instead.

The process400then determines (at step415) whether the requested data is stored in a local cache memory (e.g., the cache memory114of the application server110). If it is determined that the requested data is stored in the local cache memory, the process400retrieves (at step420) the data value corresponding to the request for data from the local cache memory and provides (at step440) the requested data to the requesting device.

On the other hand, if it is determined that the requested data is not stored in the local cache, the process400retrieves (at step425) the data value corresponding to the request for data from the local database. For example, the application server110may generate a database query (e.g., SQL query) based on the request for data, and may run the query against the local database to obtain a result set comprising the requested data. Upon retrieving the requested data from the local database, the application server110may store (at step430) the retrieved data in the local cache memory (e.g., in the cache memory114) and may update (at step435) the cache index (e.g., the cache index130) to reflect that the data value corresponding to the request for data is now cached locally at the application server110. For example, the application server110may insert a new entry in the cache index130. The new entry may include a key corresponding to the requested data, a network address corresponding to the application server110, and may also include a date and time when the data is stored in the cache memory. The application server110may also provide (at step440) the retrieved data to the requesting device.

FIG. 5is a block diagram of a computer system500suitable for implementing one or more embodiments of the present disclosure, including the application servers (e.g., the application servers110and120) of the database system104, and the requesting device102. In various implementations, the requesting device102may include a mobile cellular phone, personal computer (PC), laptop, wearable computing device, etc. adapted for wireless communication, and each of the application servers110and120may include a network computing device, such as a server. Thus, it should be appreciated that the devices110,120, and102may be implemented as computer system500in a manner as follows.

Computer system500includes a bus512or other communication mechanism for communicating information data, signals, and information between various components of computer system500. Components include an input/output (I/O) component504that processes a user (i.e., sender, recipient, service provider) action, such as selecting keys from a keypad/keyboard, selecting one or more buttons or links, etc., and sends a corresponding signal to bus512. I/O component504may also include an output component, such as a display502and a cursor control508(such as a keyboard, keypad, mouse, etc.). The display502may be configured to present a login page for logging into a user account or a checkout page for purchasing an item from a merchant associated with the database system104. An optional audio input/output component506may also be included to allow a user to use voice for inputting information by converting audio signals. Audio I/O component506may allow the user to hear audio. A transceiver or network interface520transmits and receives signals between computer system500and other devices, such as another user device, a merchant server, or a service provider server via network522. In one embodiment, the transmission is wireless, although other transmission mediums and methods may also be suitable. A processor514, which can be a micro-controller, digital signal processor (DSP), or other processing component, processes these various signals, such as for display on computer system500or transmission to other devices via a communication link524. Processor514may also control transmission of information, such as cookies or IP addresses, to other devices.

Components of computer system500also include a system memory component510(e.g., RAM), a static storage component516(e.g., ROM), and/or a disk drive518(e.g., a solid state drive, a hard drive). Computer system500performs specific operations by processor514and other components by executing one or more sequences of instructions contained in system memory component510. For example, processor514can receive a request for data from a requesting device via a network, process the request for data according to the process300or the process400, and transmit the requested data back to the requesting device via the network.

Logic may be encoded in a computer readable medium, which may refer to any medium that participates in providing instructions to processor514for execution. Such a medium may take many forms, including but not limited to, non-volatile media, volatile media, and transmission media. In various implementations, non-volatile media includes optical or magnetic disks, volatile media includes dynamic memory, such as system memory component510, and transmission media includes coaxial cables, copper wire, and fiber optics, including wires that comprise bus512. In one embodiment, the logic is encoded in non-transitory computer readable medium. In one example, transmission media may take the form of acoustic or light waves, such as those generated during radio wave, optical, and infrared data communications.

In various embodiments of the present disclosure, execution of instruction sequences to practice the present disclosure may be performed by computer system400. In various other embodiments of the present disclosure, a plurality of computer systems400coupled by communication link524to the network (e.g., such as a LAN, WLAN, PTSN, and/or various other wired or wireless networks, including telecommunications, mobile, and cellular phone networks) may perform instruction sequences to practice the present disclosure in coordination with one another.