pandaserver/taskbuffer/GlobalShares.py

0001 # Definitions
0002 EXECUTING = "executing"
0003 QUEUED = "queued"
0004 PLEDGED = "pledged"
0005
0006
0007 class Node(object):
0008     def __init__(self):
0009         self.children = []
0010
0011     def add_child(self, node):
0012         self.children.append(node)
0013
0014     def get_leaves(self, leaves=[]):
0015         # If the node has no leaves, return the node in a list
0016         if not self.children:
0017             leaves.append(self)
0018             return leaves
0019
0020         # Recursively get to the bottom
0021         for child in self.children:
0022             child.get_leaves(leaves)
0023
0024         return leaves
0025
0026
0027 class Share(Node):
0028     """
0029     Implement the share node
0030     """
0031
0032     _attributes = (
0033         "name",
0034         "value",
0035         "parent",
0036         "prodsourcelabel",
0037         "workinggroup",
0038         "campaign",
0039         "processingtype",
0040         "transpath",
0041         "vo",
0042         "rtype",
0043         "queue_id",
0044         "throttled",
0045     )
0046
0047     def __str__(self, level=0):
0048         """
0049         Print the tree structure
0050         """
0051         ret = "{0} name: {1}, value: {2}\n".format("\t" * level, self.name, self.value)
0052         for child in self.children:
0053             ret += child.__str__(level + 1)
0054         return ret
0055
0056     def __repr__(self):
0057         return self.__str__()
0058
0059     def __mul__(self, other):
0060         """
0061         If I multiply a share object by a number, multiply the value field
0062         """
0063         self.value *= other
0064         return self.value
0065
0066     def __rmul__(self, other):
0067         return self.__mul__
0068
0069     def __imul__(self, other):
0070         return self.__mul__
0071
0072     def __init__(
0073         self,
0074         name,
0075         value,
0076         parent,
0077         prodsourcelabel,
0078         workinggroup,
0079         campaign,
0080         processingtype,
0081         transpath,
0082         rtype,
0083         vo,
0084         queue_id,
0085         throttled,
0086     ):
0087         # Create default attributes
0088         for attr in self._attributes:
0089             setattr(self, attr, None)
0090
0091         Node.__init__(self)
0092         self.name = name
0093         self.value = value
0094         self.parent = parent
0095         self.prodsourcelabel = prodsourcelabel
0096         self.workinggroup = workinggroup
0097         self.campaign = campaign
0098         self.processingtype = processingtype
0099         self.transpath = transpath
0100         self.rtype = rtype
0101         self.vo = vo
0102         self.queue_id = queue_id
0103         self.throttled = throttled
0104
0105     def pretty_print_hs_distribution(self, hs_distribution, level=0):
0106         try:
0107             executing = hs_distribution[self.name][EXECUTING] / 1000.0
0108         except Exception:
0109             executing = 0
0110
0111         try:
0112             target = hs_distribution[self.name][PLEDGED] / 1000.0
0113         except Exception:
0114             target = 0
0115
0116         try:
0117             queued = hs_distribution[self.name][QUEUED] / 1000.0
0118         except Exception:
0119             queued = 0
0120
0121         ret = "{0} name: {1}, values: {2:.1f}k|{3:.1f}k|{4:.1f}k\n".format("\t" * level, self.name, executing, target, queued)
0122         for child in self.children:
0123             ret += child.pretty_print_hs_distribution(hs_distribution, level + 1)
0124         return ret
0125
0126     def normalize(self, multiplier=100, divider=100):
0127         """
0128         Will run down the branch and normalize values beneath
0129         """
0130         self.value *= multiplier * 1.0 / divider
0131         if not self.children:
0132             return
0133
0134         divider = 0
0135         for child in self.children:
0136             divider += child.value
0137
0138         multiplier = self.value
0139
0140         for child in self.children:
0141             child.normalize(multiplier=multiplier, divider=divider)
0142
0143         return
0144
0145     def sort_branch_by_current_hs_distribution(self, hs_distribution):
0146         """
0147         Runs down the branch in order of under-pledging. It returns a list of sorted leave shares
0148         """
0149         sorted_shares = []
0150
0151         # If the node has no leaves, return the node in a list
0152         if not self.children:
0153             sorted_shares = [self]
0154             return sorted_shares
0155
0156         # If the node has leaves, sort the children
0157         children_sorted = []
0158         for child1 in self.children:
0159             loop_index = 0
0160             insert_index = len(children_sorted)  # insert at the end, if not deemed otherwise
0161
0162             # Calculate under-pledging
0163             try:
0164                 child1_under_pledge = hs_distribution[child1.name][EXECUTING] * 1.0 / hs_distribution[child1.name][PLEDGED]
0165             except ZeroDivisionError:
0166                 child1_under_pledge = 10**6  # Initialize to a large default number
0167
0168             for child2 in children_sorted:
0169                 try:
0170                     # Calculate under-pledging
0171                     child2_under_pledge = hs_distribution[child2.name][EXECUTING] * 1.0 / hs_distribution[child2.name][PLEDGED]
0172                 except ZeroDivisionError:
0173                     child2_under_pledge = 10**6  # Initialize to a large default number
0174                 except KeyError:
0175                     continue  # Does not exist
0176
0177                 if child1_under_pledge < child2_under_pledge:
0178                     insert_index = loop_index
0179                     break
0180
0181                 loop_index += 1
0182
0183             # Insert the child into the list
0184             children_sorted.insert(insert_index, child1)
0185
0186         # Go recursively and sort the grand* children
0187         for child in children_sorted:
0188             sorted_shares.extend(child.sort_branch_by_current_hs_distribution(hs_distribution))
0189
0190         return sorted_shares
0191
0192     def aggregate_hs_distribution(self, hs_distribution):
0193         """
0194         We have the current HS distribution values for the leaves, but want to propagate it updwards to the parents.
0195         We will traverse the tree from top to bottom and bring up the aggregated values.
0196         """
0197         executing, queued, pledged = 0, 0, 0
0198
0199         # If the node has no children, it's a leave and should have an entry in the hs_distribution
0200         if not self.children:
0201             try:
0202                 executing = hs_distribution[self.name][EXECUTING]
0203                 queued = hs_distribution[self.name][QUEUED]
0204                 pledged = hs_distribution[self.name][PLEDGED]
0205             except KeyError:
0206                 pass
0207
0208             return executing, queued, pledged
0209
0210         # If the node has children, sum up the values of the children
0211         executing = 0
0212         queued = 0
0213         pledged = 0
0214
0215         for child in self.children:
0216             (
0217                 executing_child,
0218                 queued_child,
0219                 pledged_child,
0220             ) = child.aggregate_hs_distribution(hs_distribution)
0221             executing += executing_child
0222             queued += queued_child
0223             pledged += pledged_child
0224
0225         # Add the aggregated value to the map
0226         hs_distribution[self.name] = {
0227             EXECUTING: executing,
0228             QUEUED: queued,
0229             PLEDGED: pledged,
0230         }
0231
0232         # Return the aggregated values
0233         return executing, queued, pledged
0234
0235     # return column names
0236     def column_names(cls):
0237         ret = ""
0238         for attr in cls._attributes:
0239             if ret != "":
0240                 ret += ","
0241             ret += attr
0242         return ret
0243
0244     column_names = classmethod(column_names)