# US 10,310,748 B2 · Determining data locality in a distributed system using aggregation of locality summaries > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 10,310,748 B2 - Document ID: US10310748B2 - Status: Granted - Kind: B2 - Application number: 15/249,138 - Filed: 2016-08-26 - Published: 2017-03-02 - Granted: 2019-06-04 - Earliest claimed: 2015-08-26 - Assignees: Pivotal Software, Inc. - Applicants: Pivotal Software, Inc. - Inventors: Harshad Deshmukh; Adalbert Gerald Soosai Raj; Jignesh M. Patel - Web page: https://patentgrep.com/patent/US10310748B2 ## CPC - G06F 3/0619: {in relation to data integrity, e.g. data losses, bit errors} - G06F 3/0617: {in relation to availability} - G06F 3/065: {Replication mechanisms} - G06F 3/0653: {Monitoring storage devices or systems} - G06F 3/067: {Distributed or networked storage systems, e.g. storage area networks [SAN], network attached storage [NAS]} - G06F 16/2255: {Hash tables} - G06F 16/244: {Grouping and aggregation} - G06F 16/9535: Search customisation based on user profiles and personalisation - G06F 16/9538: Presentation of query results - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract This specification describes methods, systems, and computer program products for maintaining data representing where each data block of multiple data blocks are stored among multiple computing nodes. Each computing node generates a respective locality summary based on locally stored data blocks, and submits the locality summary to a controlling computing node. ## Claims ### Claim 1 A computer-implemented method comprising: storing locally in main memory of each computing node of a plurality of computing nodes, by each respective computing node of the plurality of computing nodes, respective data blocks, wherein each data block is a piece of data, wherein each data block is associated with a respective block identifier that is a unique identifier over all data blocks stored on all computing nodes, and each data block is one of a plurality of data blocks; determining, by each computing node of the plurality of computing nodes, one or more block identifiers of data blocks stored locally in main memory of the computing node; generating, by each computing node of the plurality of computing nodes, a respective locality summary having n positions and n corresponding values, wherein n is an integer less than a total number of the data blocks, wherein the respective locality summary represents which of the plurality of data blocks are stored locally by the computing node, and wherein generating the locality summary includes mapping each of the one or more block identifiers stored by the computing node to a respective position in the locality summary and setting a value at the respective position in the locality summary; and sending, by each computing node of the plurality of computing nodes to a controller computing node, the respective locality summary generated by the computing node. ### Claim 2 (depends on 1) The method of claim 1, wherein mapping each of the one or more block identifiers to the respective position in the locality summary comprises computing a position index identified by (i mod n), wherein i is a block identifier, i being an integer, and n is a size of the locality summary. ### Claim 3 (depends on 1) The method of claim 1, further comprising: periodically regenerating, by each computing node of the plurality of computing nodes, the locality summary for the computing node. ### Claim 4 (depends on 1) The method of claim 1, wherein the locality summary is implemented as a Bloom filter. ### Claim 5 (depends on 1) The method of claim 1, further comprising: receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; determining, by the controller computing node from the locality summaries sent to the controller computing node, which computing nodes are candidate computing nodes, candidate computing nodes being computing nodes that are not excluded by the locality summaries as nodes that are storing the particular block; and requesting, from each candidate computing node, data indicating whether or not the candidate computing node locally stores the particular data block. ### Claim 6 A computer-implemented method comprising: receiving, by a controller computing node, locality summaries from a plurality of computing nodes, wherein each locality summary identifies which of a plurality of data blocks are stored locally on a respective computing node by mapping each of one or more block identifiers identifying a respective locally stored data block to a respective position in the locality summary and setting a value at the respective position in the locality summary; receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; mapping the particular block identifier to a particular position in received locality summaries; determining, by the controller computing node, which of the particular computing nodes provided a locality summary having a value set at the particular position; and requesting, from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block. ### Claim 7 (depends on 6) The method of claim 6, further comprising: receiving, by the controller computing node, an indication that a first computing node of the particular computing nodes locally stores the particular data block; and providing, by the controller computing node and to the first computing node, a request to perform a computation using the particular data block. ### Claim 8 (depends on 6) The method of claim 6, further comprising: maintaining, by the controller computing node, a respective bit vector for each position in the locality summaries received from the plurality of computing nodes, wherein each position in the bit vector represents whether a particular node stores a data block corresponding to the position in the locality summaries, wherein determining which of the particular computing nodes provided a locality summary having a value set at the particular position comprises identifying nonzero positions in the bit vector. ### Claim 9 (depends on 8) The method of claim 8, further comprising: maintaining, by the controller computing node, a mapping between computing node identifiers and positions in the bit vector; and mapping the nonzero positions in the bit vector to respective computing node identifiers. ### Claim 10 (depends on 6) The method of claim 6, further comprising: generating, by the controller computing node and for a plurality of locality summaries received over time from the plurality of computing nodes, a respective likelihood that each computing node of the plurality of computing nodes currently stores a particular data block. ### Claim 11 (depends on 10) The method of claim 10, wherein requesting, from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block comprises requesting the data from the computing nodes in an order determined by the respective likelihoods that that the particular computing nodes currently store the particular data block. ### Claim 12 A system comprising: one or more computers and one or more storage devices storing instructions that are operable, when executed by the one or more computers, to cause the one or more computers to perform operations comprising: storing locally in main memory of each computing node of a plurality of computing nodes, by each respective computing node of the plurality of computing nodes, respective data blocks, wherein each data block is a piece of data, wherein each data block is associated with a respective block identifier that is a unique identifier over all data blocks stored on all computing nodes, and each data block is one of a plurality of data blocks; determining, by each computing node of the plurality of computing nodes, one or more block identifiers of data blocks stored locally in main memory of the computing node; generating, by each computing node of the plurality of computing nodes, a respective locality summary having n positions and n corresponding values, wherein n is an integer less than a total number of the data blocks, wherein the respective locality summary represents which of the plurality of data blocks are stored locally by the computing node, and wherein generating the locality summary includes mapping each of the one or more block identifiers stored by the computing node to a respective position in the locality summary and setting a value at the respective position in the locality summary; and sending, by each computing node of the plurality of computing nodes to a controller computing node, the respective locality summary generated by the computing node. ### Claim 13 (depends on 12) The system of claim 12, wherein mapping each of the one or more block identifiers to the respective position in the locality summary comprises computing a position index identified by (i mod n), wherein i is a block identifier, i being an integer, and n is a size of the locality summary. ### Claim 14 (depends on 12) The system of claim 12, wherein the operations further comprise: periodically regenerating, by each computing node of the plurality of computing nodes, the locality summary for the computing node. ### Claim 15 (depends on 12) The system of claim 12, wherein the locality summary is implemented as a Bloom filter. ### Claim 16 (depends on 12) The system of claim 12, wherein the system further comprises the controller computing node and wherein the controller computing node comprises one or more controller computers and one or more controller storage devices storing instructions that are operable, when executed by the one or more controller computers to perform operations comprising: receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; determining, by the controller computing node from the locality summaries sent to the controller computing node, which computing nodes are candidate computing nodes, candidate computing nodes being computing nodes that are not excluded by the locality summaries as nodes that are storing the particular block; and requesting, from each candidate computing node, data indicating whether or not the candidate computing node locally stores the particular data block. ### Claim 17 A system comprising: one or more computers and one or more storage devices storing instructions that are operable, when executed by the one or more computers, to cause the one or more computers to perform operations comprising: receiving, by a controller computing node, locality summaries from a plurality of computing nodes, wherein each locality summary identifies which of a plurality of data blocks are stored locally on a respective computing node by mapping each of one or more block identifiers identifying a respective locally stored data block to a respective position in the locality summary and setting a value at the respective position in the locality summary; receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; mapping, by the controller computing node, the particular block identifier to a particular position in received locality summaries; determining, by the controller computing node, which of the particular computing nodes provided a locality summary having a value set at the particular position; and requesting, by the controller computing node and from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block. ### Claim 18 (depends on 17) The system of claim 17, wherein the operations further comprise: receiving, by the controller computing node, an indication that a first computing node of the particular computing nodes locally stores the particular data block; and providing, by the controller computing node and to the first computing node, a request to perform a computation using the particular data block. ### Claim 19 (depends on 17) The system of claim 17, wherein the operations further comprise: maintaining, by the controller computing node, a respective bit vector for each position in the locality summaries received from the plurality of computing nodes, wherein each position in the bit vector represents whether a particular node stores a data block corresponding to the position in the locality summaries, wherein determining which of the particular computing nodes provided a locality summary having a value set at the particular position comprises identifying nonzero positions in the bit vector. ### Claim 20 (depends on 19) The system of claim 19, wherein the operations further comprise: maintaining, by the controller computing node, a mapping between computing node identifiers and positions in the bit vector; and mapping the nonzero positions in the bit vector to respective computing node identifiers. ### Claim 21 (depends on 17) The system of claim 17, wherein the operations further comprise: generating, by the computing node and for a plurality of locality summaries received over time from the plurality of computing nodes, a respective likelihood that each computing node of the plurality of computing nodes currently stores a particular data block. ### Claim 22 (depends on 21) The system of claim 21, wherein requesting, by the computing node and from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block comprises requesting the data from the computing nodes in an order determined by the respective likelihoods that that the particular computing nodes currently store the particular data block. ### Claim 23 A computer program product, encoded on one or more non-transitory computer storage media, comprising instructions that when executed by one or more computers cause the one or more computers to perform operations comprising: storing locally in main memory of each computing node of a plurality of computing nodes, by each respective computing node of the plurality of computing nodes, respective data blocks, wherein each data block is a piece of data, wherein each data block is associated with a respective block identifier that is a unique identifier over all data blocks stored on all computing nodes, and each data block is one of a plurality of data blocks determining, by each computing node of the plurality of computing nodes, one or more block identifiers of data blocks stored locally in main memory of the computing node; generating, by each computing node of the plurality of computing nodes, a respective locality summary having n positions and n corresponding values, wherein n is an integer less than a total number of the data blocks, wherein the respective locality summary represents which of the plurality of data blocks are stored locally by the computing node, and wherein generating the locality summary includes mapping each of the one or more block identifiers stored by the computing node to a respective position in the locality summary and setting a value at the respective position in the locality summary; and sending, by each computing node of the plurality of computing nodes to a controller computing node, the respective locality summary generated by the computing node. ### Claim 24 (depends on 23) The computer program product of claim 23, wherein mapping each of the one or more block identifiers to the respective position in the locality summary comprises computing a position index identified by (i mod n), wherein i is a block identifier, i being an integer, and n is a size of the locality summary. ### Claim 25 (depends on 23) The computer program product of claim 23, wherein the operations further comprise: periodically regenerating, by each computing node of the plurality of computing nodes, the locality summary for the computing node. ### Claim 26 (depends on 23) The computer program product of claim 23, wherein the locality summary is implemented as a Bloom filter. ### Claim 27 (depends on 23) The computer program product of claim 23, wherein the operations further comprise: receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; determining, by the controller computing node from the locality summaries sent to the controller computing node, which computing nodes are candidate computing nodes, candidate computing nodes being computing nodes that are not excluded by the locality summaries as nodes that are storing the particular block; and requesting, from each candidate computing node, data indicating whether or not the candidate computing node locally stores the particular data block. ### Claim 28 A computer program product, encoded on one or more non-transitory computer storage media, comprising instructions that when executed by one or more computers cause the one or more computers to perform operations comprising: receiving, by a controller computing node, locality summaries from a plurality of computing nodes, wherein each locality summary identifies which of a plurality of data blocks are stored locally on a respective computing node by mapping each of one or more block identifiers identifying a respective locally stored data block to a respective position in the locality summary and setting a value at the respective position in the locality summary; receiving, by the controller computing node, a request to identify which computing node locally stores a particular data block identified by a particular block identifier; mapping the particular block identifier to a particular position in received locality summaries; determining, by the controller computing node, which of the particular computing nodes provided a locality summary having a value set at the particular position; and requesting, from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block. ### Claim 29 (depends on 28) The computer program product of claim 28, wherein the operations further comprise: receiving, by the controller computing node, an indication that a first computing node of the particular computing nodes locally stores the particular data block; and providing, by the controller computing node and to the first computing node, a request to perform a computation using the particular data block. ### Claim 30 (depends on 28) The computer program product of claim 28, wherein the operations further comprise: maintaining, by the controller computing node, a respective bit vector for each position in the locality summaries received from the plurality of computing nodes, wherein each position in the bit vector represents whether a particular node stores a data block corresponding to the position in the locality summaries, wherein determining which of the particular computing nodes provided a locality summary having a value set at the particular position comprises identifying nonzero positions in the bit vector. ### Claim 31 (depends on 30) The computer program product of claim 30, wherein the operations further comprise: maintaining, by the controller computing node, a mapping between computing node identifiers and positions in the bit vector; and mapping, by the controller computing node, the nonzero positions in the bit vector to respective computing node identifiers. ### Claim 32 (depends on 28) The computer program product of claim 28, wherein the operations further comprise: generating, by the controller computing node and for a plurality of locality summaries received over time from the plurality of computing nodes, a respective likelihood that each computing node of the plurality of computing nodes currently stores a particular data block. ### Claim 33 (depends on 32) The computer program product of claim 32, wherein requesting, by the controller computing node and from each computing node of the particular computing nodes, data indicating whether or not the computing node locally stores the particular data block comprises requesting the data from the computing nodes in an order determined by the respective likelihoods that that the particular computing nodes currently store the particular data block.