Copysets/RepeatedFailures.py at master · jaynavar/Copysets · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import collections
import itertools as it
import numpy as np
import random

class Runner(object):
   def __init__(self, numNodes, scatterWidth, failureInterval, numIntervals,
                numTrials, replicationFactor, nodeBandwidth, nodeCapacity,
                recoveryUtil):
      self.numNodes = numNodes
      self.scatterWidth = scatterWidth
      self.failureInterval = failureInterval
      self.numIntervals = numIntervals
      self.numTrials = numTrials
      self.replicationFactor = replicationFactor
      self.nodeBandwidth = nodeBandwidth
      self.nodeCapacity = nodeCapacity
      self.recoveryUtil = recoveryUtil

      # computed params
      self.permutations = int(self.scatterWidth / float(self.replicationFactor - 1))

      # simulated cluster state
      self.copysets = None
      self.buddies = None
      self.liveNodes = None
      self.failedNodes = None
      self.lostData = None

   def run(self):
      isolatedProbs = [[] for _ in range(self.numIntervals)]
      compoundingProbs = [[] for _ in range(self.numIntervals)]
      for _ in range(self.numTrials):
         # setup the cluster
         self.setup()
         for interval in range(self.numIntervals):
            # cause repeated failure, and compute probability of data loss
            isolatedProbs[interval].append(self.failureProbOfDataLoss())
            compoundingProbs[interval].append(1.0 if self.lostData else 0.0)
            # simulate recovery of the cluster over given interval
            # before next failure
            self.recover()

      # average the results from the trials
      data = []
      for interval in range(self.numIntervals):
         data.append((interval * self.failureInterval,
                      (np.array(isolatedProbs[interval]).mean(),
                       np.array(compoundingProbs[interval]).mean())))
      return data

   def setup(self):
      # simulated cluster state
      self.copysets = set()
      self.buddies = collections.defaultdict(set)
      self.liveNodes = set(range(self.numNodes))
      self.failedNodes = set()
      self.lostData = False

      # generate copysets using Copyset Replication scheme
      shuffledNodes = range(self.numNodes)
      for p in xrange(self.permutations):
         # permute the nodes
         random.shuffle(shuffledNodes)
         # separate them into copysets, add to set of all copysets
         for i in xrange(0, len(shuffledNodes), self.replicationFactor):
            copyset = tuple(
               sorted(shuffledNodes[i : i + self.replicationFactor]))
            if len(copyset) == 3:
               self.copysets.add(copyset)

      # create mapping from node to the other nodes it shares
      # copysets with, used to determine recovery time
      for copyset in self.copysets:
         for node in copyset:
            self.buddies[node].update(copyset)
      for node in range(self.numNodes):
         self.buddies[node].remove(node)

   def failureProbOfDataLoss(self):
      # fail 1% of the nodes (remove from live, add to failed)
      newFailedNodes = set(
         random.sample(self.liveNodes, int(0.01 * len(self.liveNodes))))
      self.liveNodes.difference_update(newFailedNodes)
      self.failedNodes.update(newFailedNodes)

      # determine if failed nodes form one of the generated copysets
      lostData = not self.copysets.isdisjoint(
         it.combinations(sorted(self.failedNodes), self.replicationFactor))

      if lostData:
         self.lostData = True
         return 1.0
      else:
         return 0.0

   def recover(self):
      originalFailedNodes = self.failedNodes.copy()

      failedBuddies = collections.defaultdict(set)
      for failedNode in originalFailedNodes:
         # create mapping from alive nodes to nodes
         # they are helping recover
         aliveBuddies = [buddy for buddy in self.buddies[failedNode]
                         if buddy not in originalFailedNodes]
         for aliveBuddy in aliveBuddies:
            failedBuddies[aliveBuddy].add(failedNode)

      for failedNode in originalFailedNodes:
         aliveBuddies = [buddy for buddy in self.buddies[failedNode]
                         if buddy not in originalFailedNodes]

         # determine recovery time, based on network bandwidth of recovering node,
         # bandwidth of each peer that can be dedicated to recovery of this node,
         # and amount of data being recovered
         totalBandwidth = min(self.nodeBandwidth, sum(
            [self.recoveryUtil * self.nodeBandwidth /
             float(len(failedBuddies[aliveBuddy]))
             for aliveBuddy in aliveBuddies]))
         recoveryTime = self.nodeCapacity / float(totalBandwidth)

         # check if recovered before next failure
         if recoveryTime < self.failureInterval:
            # remove from failed set, add to live set
            self.failedNodes.remove(failedNode)
            self.liveNodes.add(failedNode)