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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
|
########################################################################################
# #
# Author: Bertrand Neron, #
# Organization:'Biological Software and Databases' Group, Institut Pasteur, Paris. #
# Distributed under GPLv2 Licence. Please refer to the COPYING.LIB document. #
# #
########################################################################################
import errno
from lxml import etree
import fcntl
from time import sleep , time
from Mobyle.Status import Status
from Mobyle.MobyleError import MobyleError
#from Mobyle.Utils import parse_xml_file
import logging
_log = logging.getLogger( __name__ )
import sys
import os.path
class StatusManager(object):
"""
This class is the main access to the job status
"""
WRITE = fcntl.LOCK_EX
READ = fcntl.LOCK_SH
file_name = 'mobyle_status.xml'
@classmethod
def create( cls , job_path , status ):
fileName = os.path.join( job_path , cls.file_name )
root = etree.Element( "status" )
value_node = etree.Element( "value" )
root.append( value_node )
message_node = etree.Element( "message" )
root.append( message_node )
value_node.text = str( status )
message_node.text = status.message
try:
File = open( fileName , 'w' )
File.write( etree.tostring( root , encoding='UTF-8' , pretty_print = True))
except IOError , err:
msg = "cannot create status %s : %s" % ( fileName , err )
_log.error( msg )
raise MobyleError( msg )
finally:
File.close()
def _lock( self , File , lockType ):
"""
try to acquire a lock of lockType on self.__File
@raise IOError: when it could not acquire a lock
"""
IGotALock = False
_log.debug( "%f : %s : _lock Type= %s ( call by= %s )" %( time() ,
File.name,
( 'UNKNOWN LOCK', 'READ' , 'WRITE' )[ lockType ] ,
os.path.basename( sys.argv[0] ) ,
))
for attempt in range( 5 ):
try:
fcntl.lockf( File , lockType | fcntl.LOCK_NB )
IGotALock = True
_log.debug( "%f : %s : _lock IGotALock = True" %(time() , File.name ))
break
except IOError , err:
_log.debug( "%f : %s : _lock IGotALock = False" %(time() , File.name))
sleep( 0.2 )
if not IGotALock :
_log.error( "%s : %s" %( File.name , err ) )
_log.debug( "%f : %s : _lock Type= %s ( call by= %s )" %( time() ,
File.name ,
( 'UNKNOWN LOCK', 'READ' , 'WRITE' )[ lockType ] ,
os.path.basename( sys.argv[0] )
))
raise IOError , err
def setStatus(self, job_path , status ):
fileName = os.path.normpath( os.path.join( job_path , self.file_name ) )
try:
File = open( fileName , 'r+' )
except IOError , err:
if err.errno == errno.ENOENT :
try:
return self._fall_back_setStatus( job_path , status )
except Exception , err:
_log.error( "%s : cannot open either index.xml nor mobyle_status.xml: %s" % ( job_path , err ))
return Status( code= -1 )
else:
_log.error( "cannot open %s : %s" % ( fileName , err ))
return Status( code= -1 )
self._lock( File , self.WRITE ) #try to acquire a lock
#If I do not got a lock a IOError is raised
try:
parser = etree.XMLParser( no_network = False )
#doc = parse_xml_file(File , parser)
doc = etree.parse(File , parser)
root = doc.getroot()
except Exception , err:
msg = "error in parsing status %s : %s" % ( File.name , err )
_log.error( msg )
_log.debug("%f : %s : setStatus UNLOCK " %( time() , fileName ) )
fcntl.lockf( File , fcntl.LOCK_UN )
File.close()
raise MobyleError( msg )
#################
old_status = self.getStatus( job_path )
_log.debug( "old_status= %s" %old_status )
if old_status.isEnded():
_log.warning( "%f : %s : try to update job status from %s to %s ( call by %s )"%( time() ,
fileName ,
old_status ,
status ,
os.path.basename( sys.argv[0] )
) )
try:
_log.debug("%f : %s : setStatus UNLOCK " %( time() , fileName ) )
fcntl.lockf( File , fcntl.LOCK_UN )
except Exception , err :
_log.error( "%f : %s : setStatus cannot UNLOCK : %s" %( time() , fileName , err ) )
finally:
File.close()
return None
try:
value_node = root.find( "value" )
value_node.text = str( status )
message_node = root.find( "message" )
message_node.text = status.message
except Exception, err:
msg = "error in parsing status %s : %s" % ( File.name , err )
_log.error( msg )
_log.debug("%f : %s : setStatus UNLOCK " %( time() , fileName ) )
fcntl.lockf( File , fcntl.LOCK_UN )
File.close()
raise MobyleError( msg )
##################
try:
tmpFile = open( "%s.%d" %( File.name , os.getpid() ) , 'w' )
tmpFile.write( etree.tostring( root , xml_declaration=True , encoding='UTF-8' , pretty_print = True))
os.rename( tmpFile.name , File.name )
except IOError , err:
msg = "cannot write status : " + str( err )
_log.error( msg )
raise MobyleError( msg )
except Exception , err:
_log.error( "cannot write status %s " % File.name , exc_info = True )
raise err
finally:
try:
_log.debug("%f : %s : setStatus UNLOCK " %( time() , fileName ) )
fcntl.lockf( File , fcntl.LOCK_UN )
File.close()
tmpFile.close()
except Exception , err :
_log.error( "%f : %s : setStatus cannot UNLOCK : %s" %( time() , fileName , err ) )
def getStatus(self , job_path ):
fileName = os.path.normpath( os.path.join( job_path , self.file_name ) )
try:
File = open( fileName , 'r' )
except IOError , err:
if err.errno == errno.ENOENT :
try:
return self._fall_back_getStatus( job_path )
except Exception , err:
_log.error( "%s : cannot open either index.xml nor mobyle_status.xml: %s" % ( job_path , err ))
return Status( code= -1 )
else:
_log.error( "cannot open %s : %s" % ( fileName , err ))
return Status( code= -1 )
try:
self._lock( File, self.READ ) #try to acquire a lock
#If I do not got a lock a IOError is raised
except IOError , err:
_log.error( "%f : %s : cannot read status : %s" %( time() , fileName , err ) )
return Status( code= -1 )
try:
parser = etree.XMLParser( no_network = False )
#doc = parse_xml_file( File , parser )
doc = etree.parse(File, parser)
root = doc.getroot()
except Exception , err:
pass
finally:
_log.debug("%f : %s : getStatus UNLOCK " %( time() , fileName ) )
fcntl.lockf( File , fcntl.LOCK_UN )
File.close()
try:
value_node = root.find( "value" )
message_node = root.find( "message" )
except Exception , err:
_log.error( "%f : %s : cannot parse status" %( time() , fileName ) , exc_info = True )
return Status(code = -1)
if message_node is None :
return Status( string = value_node.text )
else:
return Status( string = value_node.text , message = message_node.text )
def _fall_back_setStatus(self , job_path , status ):
parser = etree.XMLParser( no_network = False )
index_path = os.path.join( job_path ,'index.xml')
#doc = parse_xml_file( index_path , parser)
doc = etree.parse(index_path, parser)
root = doc.getroot()
status_node = root.find( "./status" )
try:
old_message = status_node.find( "value" )
if old_message is None:
oldStatus = Status( string = str( status_node.find( "value").text ) )
else:
oldStatus = Status( string = str( status_node.find( "value").text ) ,
message = str( old_message.text )
)
except AttributeError:
oldStatus = Status( code = -1 ) #unknown
if oldStatus.isKnown() and oldStatus.isEnded():
#the job is ended we can refactor status in a separate file
root.remove( status_node )
try:
tmp_file = open( os.path.join( job_path ,'tmp_index.xml' ) , 'w' )
tmp_file.write( etree.tostring( doc , pretty_print = True , encoding='UTF-8' ))
tmp_file.close()
os.rename( tmp_file.name , index_path )
except Exception , err :
_log.error( 'cannot make index.xml : %s' % err )
self.create( job_path , oldStatus )
return None
if status == oldStatus:
#this is an old job with status managed by JobState
#new status and old status will evolved
#do nothing until job finish
return None
else:
if status.isEnded():
#the job is ended we can refactor status in a separate file
root.remove( status_node )
try:
tmp_file = open( os.path.join( job_path ,'tmp_index.xml' ) , 'w' )
tmp_file.write( etree.tostring( doc , pretty_print = True , encoding='UTF-8' ))
tmp_file.close()
os.rename( tmp_file.name , index_path )
except Exception , err :
_log.error( 'cannot make index.xml : %s' % err )
self.create( job_path , status)
else:
#the job is not ended we update the index.xml
value_node = status_node.find( 'value' )
value_node.text = str( status )
message_node = status_node.find( 'message' )
if message_node is not None:
message_node.text = str( status.message )
try:
tmpFile = open( "%s.%d" %( index_path , os.getpid() ) , 'w' )
tmpFile.write( etree.tostring( doc , xml_declaration=True , encoding='UTF-8' ))
tmpFile.close()
os.rename( tmpFile.name , index_path )
except IOError, err:
_log.error( "IOError during write index.xml on disk")
raise MobyleError , err
def _fall_back_getStatus(self , job_path ):
parser = etree.XMLParser( no_network = False )
#doc = parse_xml_file( os.path.join( job_path ,'index.xml') , parser)
doc = etree.parse(os.path.join(job_path, 'index.xml'), parser)
root = doc.getroot()
try:
status = root.find( "./status/value" ).text
except AttributeError:
return Status( code = -1 ) #unknown
try:
message = root.find( "./status/message" ).text
except AttributeError:
message = None
if message:
return Status( string = str( status ) , message = str( message ))
else:
return Status( string = str( status ) )
|
