OpenStack Swift源码导读之——业务整体架构和Proxy进程

container_partition, containers = self.app.container_ring.get_nodes( 

 self.account_name, self.container_name)
def get_nodes(self, account, container=None, obj=None): 

 """ 
 Get the partition and nodes 
for an account/container/object. 
 If a node is responsible 
for more than one replica, it will 
 only appear in the 
output once.
 :param account: account name 
 :param 
container: container name 
 :param obj: object name 

 :returns: a tuple of (partition, list of node dicts)
 Each node dict will have at least the following keys:
 ====== 
=============================================================== 

 id unique integer 
identifier amongst devices 
 weight a float of the 
relative weight of this device as compared to 

 others; 
this indicates how many partitions the builder will try 

 to assign 
to this device 
 zone integer indicating 
which zone the device is in; a given 

 partition 
will not be assigned to multiple devices within the 

 same zone 

 ip the ip address of the 
device 
 port the tcp port of the device 

 device the device's name on disk (sdb1, for 
example) 
 meta general use 'extra' 
field; for example: the online date, the 

 hardware 
description 
 ====== 
=============================================================== 

 """ 
 part = self.get_part(account, 
container, obj) 
 return part, 
self._get_part_nodes(part)
def get_part(self, account, container=None, obj=None): 
 
""" 
 Get the partition for an 
account/container/object.
 :param account: account name 
 :param 
container: container name 
 :param obj: object name 

 :returns: the partition number 
 """ 

 key = hash_path(account, container, obj, raw_digest=True) 

 if time() >; self._rtime: 

 self._reload() 

 part = struct.unpack_from('>;I', key)[0] >> 
self._part_shift 
 return part
def _get_part_nodes(self, part): 
 part_nodes = [] 

 seen_ids = set() 
 for r2p2d in 
self._replica2part2dev_id: 
 if 
part <; len(r2p2d): 

 dev_id = 
r2p2d[part] 

 if dev_id 
not in seen_ids: 

 
part_nodes.append(self.devs[dev_id]) 

 
seen_ids.add(dev_id) 
 return part_nodes

def make_requests(self, req, ring, part, method, path, headers, 

 
query_string=''): 
 """ 
 Sends an 
HTTP request to multiple nodes and aggregates the results. 

 It attempts the primary nodes concurrently, then iterates 
over the 
 handoff nodes as needed.
 :param req: a request sent by the client 

 :param ring: the ring used for finding backend servers 

 :param part: the partition number 
 
:param method: the method to send to the backend 
 :param 
path: the path to send to the backend 

 
(full path ends up being /<$device>/<$part>/<$path>) 

 :param headers: a list of dicts, where each dict 
represents one 

 
backend request that should be made. 
 :param query_string: 
optional query string to send to the backend 
 :returns: a 
swob.Response object 
 """ 
 
start_nodes = ring.get_part_nodes(part) 
 nodes = 
GreenthreadSafeIterator(self.app.iter_nodes(ring, part)) 
 
pile = GreenAsyncPile(len(start_nodes)) 
 for head in 
headers: 
 
pile.spawn(self._make_request, nodes, part, method, path, 

 
head, query_string, self.app.logger.thread_locals) 
 
response = [] 
 statuses = [] 
 for 
resp in pile: 
 if not resp: 

 continue 

 response.append(resp) 

 statuses.append(resp[0]) 

 if self.have_quorum(statuses, 
len(start_nodes)): 

 break 

 # give any pending requests *some* chance to finish 

 pile.waitall(self.app.post_quorum_timeout) 

 while len(response) <; len(start_nodes): 

 
response.append((HTTP_SERVICE_UNAVAILABLE, '', '', '')) 
 
statuses, reasons, resp_headers, bodies = zip(*response) 
 
return self.best_response(req, statuses, reasons, bodies, 

 
'%s %s' % (self.server_type, req.method), 

 
headers=resp_headers)