Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

#!/usr/bin/env python 

 

""" 

@file ion/util/task_chain.py 

@author Dave Foster <dfoster@asascience.com> 

@brief TaskChain class for sequential execution of callables (deferreds and non-deferreds) 

""" 

 

from twisted.internet import defer 

import ion.util.ionlog 

log = ion.util.ionlog.getLogger(__name__) 

 

# Disabling MutableSequence for 2.5 compat -> deriving from list for now. 

# search for MUTABLESEQUENCE to see what needs to be uncommented/fixed. 

#class TaskChain(MutableSequence): 

class TaskChain(list): 

    """ 

    Used to set up a chain of tasks that run one after another. 

 

    A task chain can be used to script a sequence of actions and have the Twisted 

    reactor manage it all. The run method returns a deferred that is called back 

    when all tasks in the chain complete. If any task errors, the chain is aborted 

    and the errback is raised. The tasks are executed in order. 

 

    The tasks should be callables that can either return deferreds or execute 

    synchronously (and TaskChain will wrap them in deferreds). If any of them error, 

    the chain is aborted and the errback is raised. 

 

    A TaskChain is a list of either callables or tuples of 2 or 3 length, that consist 

    of a callable, a list of arguments to be passed, and an optional dict of keyword 

    arguments to be passed. 

 

    Due to not having MutableSequence available in Python 2.5, type checking is done 

    only at time of execution. 

    """ 

 

    def __init__(self, *tasks): 

        """ 

        Constructor. 

 

        @param tasks  Takes a list of tasks that will be executed in order. 

        """ 

        #self._list      = []        # implementation backend for MutableSequence methods to use        # MUTABLESEQUENCE 

 

        for t in tasks: 

            self._check_type(t) 

            self.append(t) 

 

        self._donetasks = [] 

        self._results   = [] 

        self._running   = False 

        self._curtask   = None 

        self._curtask_def = None 

        self._deferred  = defer.Deferred() 

 

        self._lenprocs  = len(self) 

 

    def __str__(self): 

        """ 

        Returns a string representation of a TaskChain and its status. 

        """ 

        return "TaskChain (running=%s, %d/%d)" % (str(self._running), len(self._donetasks), len(self) + len(self._donetasks)) 

 

    def _check_type(self, obj): 

        """ 

        Internal safety mechanism that append, insert, and extend all flow through. 

        It makes sure the types being added to the list are expected. 

        """ 

        if isinstance(obj, tuple): 

            if not (len(obj) == 2 or len(obj) == 3): 

                raise ValueError("Invalid number of arguments in tuple: (callback, list of args, optional dict of kwargs) expected.") 

            if not callable(obj[0]): 

                raise ValueError("First item of tuple not a callable: (callback, list of args, optional dict of kwargs) expected.") 

            if not isinstance(obj[1], list): 

                raise ValueError("Second item of tuple not a list of args: (callback, list of args, optional dict of kwargs) expected.") 

            if len(obj) == 3 and not isinstance(obj[2], dict): 

                raise ValueError("Third item of tuple not a dict of kwargs: (callback, list of args, optional dict of kwargs) expected.") 

        else: 

            if not callable(obj): 

                raise ValueError("Item must be a callable") 

 

    # MUTABLESEQUENCE 

    #def __getitem__(self, index): 

    #    return self._list.__getitem__(index) 

 

    #def __setitem__(self, index, value): 

    #    self._check_type(value) 

    #    self._list.__setitem(index, value) 

 

    #def __delitem__(self, index): 

    #    self._list.__delitem__(index) 

 

    #def insert(self, index, value): 

    #    self._check_type(value) 

    #    self._list.insert(index, value) 

 

    #def __len__(self): 

    #    return len(self._list) 

    # END MUTABLESEQUENCE 

 

    def run(self): 

        """ 

        Starts running the chain of tasks. 

 

        @returns A deferred which will callback when the tasks complete. 

        """ 

        log.debug("TaskChain starting") 

        self._running = True 

        self._run_one() 

        return self._deferred 

 

    def _run_one(self): 

        """ 

        Runs the next task. 

        """ 

 

        # if we have no more tasks to run, or we shouldn't be running anymore, 

        # fire our callback 

        if len(self) == 0 or not self._running: 

            self._fire(True) 

            return 

 

        log.debug(self.__str__() + ":running task") 

 

        self._curtask = self.pop(0) 

        args = [] 

        kwargs = {} 

 

        # make sure this is legit - we have no way of checking on insert right now due to not being a MutableSequence 

        self._check_type(self._curtask) 

 

        if isinstance(self._curtask, tuple): 

            aslist = list(self._curtask) 

            self._curtask = aslist.pop(0) 

            args = aslist.pop(0) 

            if len(aslist): 

                kwargs = aslist.pop() 

 

        # possibly not a deferred at all! 

        self._curtask_def = defer.maybeDeferred(self._curtask, *args, **kwargs) 

        self._curtask_def.addCallbacks(self._proc_cb, self._proc_eb) 

 

    def _proc_cb(self, result): 

        """ 

        Callback on single task success. 

        """ 

        self._results.append(result) 

        self._donetasks.append(self._curtask) 

        self._curtask = None 

        self._curtask_def = None 

 

        log.debug(self.__str__() + ":task finished") 

 

        self._run_one() 

 

    def _proc_eb(self, failure): 

        """ 

        Errback on single failure. 

        """ 

        failure.trap(StandardError) 

        failure.printBriefTraceback() 

 

        log.debug(self.__str__() + ":task ERROR") 

 

        self._results.append(failure.value) 

        self._fire(False) 

 

    def _fire(self, success): 

        """ 

        Calls the task chain's callback or errback as per the success parameter. 

        This method builds the task/result list to pass back through either mechanism. 

        """ 

 

        # we're no longer running, indicate as such 

        self._running = False 

 

        log.debug(self.__str__() + ":terminating, success=%s" % str(success)) 

 

        res = zip(self._donetasks, self._results) 

        if success: 

            self._deferred.callback(res) 

        else: 

            self._deferred.errback(StandardError(res)) 

 

    def close(self): 

        """ 

        Shuts down the current chain of tasks. 

        The current executing task will have its cancel method called on its deferred. 

        It is the responsibility of the deferred creator to set up the canceller argument 

        when the deferred is constructed. 

        """ 

 

        log.debug(self.__str__() + ":close") 

 

        if not self._running: 

            # someone could call close after we've already fired, so don't fire again 

            if not self._deferred.called: 

                self._fire(True) 

            return self._deferred 

 

        self._running = False 

 

        if self._curtask_def: 

            self._curtask_def.cancel() 

 

        return self._deferred