Hide keyboard shortcuts

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#!/usr/bin/python3 

2# 

3# Copyright (C) Citrix Systems Inc. 

4# 

5# This program is free software; you can redistribute it and/or modify 

6# it under the terms of the GNU Lesser General Public License as published 

7# by the Free Software Foundation; version 2.1 only. 

8# 

9# This program is distributed in the hope that it will be useful, 

10# but WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

12# GNU Lesser General Public License for more details. 

13# 

14# You should have received a copy of the GNU Lesser General Public License 

15# along with this program; if not, write to the Free Software Foundation, Inc., 

16# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 

17 

18"""Communication for processes""" 

19 

20import os 

21import util 

22import errno 

23 

24 

25class IPCFlagException(util.SMException): 

26 pass 

27 

28 

29class IPCFlag: 

30 """Flag-based communication for processes (set, test, clear). 

31 Not thread-safe.""" 

32 

33 BASE_DIR = "/var/run/sm/ipc" 

34 

35 def __init__(self, ns): 

36 self.ns = ns 

37 self.nsDir = os.path.join(self.BASE_DIR, self.ns) 

38 if not util.pathexists(self.nsDir): 

39 try: 

40 os.makedirs(self.nsDir) 

41 except OSError: 

42 pass 

43 if not util.pathexists(self.nsDir): 

44 raise IPCFlagException("failed to create %s" % self.nsDir) 

45 

46 def set(self, name, soft=False): 

47 """Set the flag 

48 

49 name: the file to set 

50 soft: If set to False and the file gets created while this function is 

51 trying to set create, the file MAY be overwritten. 

52 

53 returns: True if the file is written, False otherwise.""" 

54 if not soft and self.test(name): # XXX this is broken! 

55 return 

56 flagFile = os.path.join(self.nsDir, name) 

57 try: 

58 if soft: 

59 f = util.open_atomic(flagFile, "w") 

60 else: 

61 f = open(flagFile, "w") 

62 f.write("%s\n" % os.getpid()) 

63 f.close() 

64 util.SMlog("IPCFlag: set %s:%s" % (self.ns, name)) 

65 return True 

66 except OSError as e: 

67 if e.errno == errno.EEXIST and soft: 

68 return False 

69 else: 

70 raise IPCFlagException("failed to create %s: %s" \ 

71 % (flagFile, e)) 

72 except IOError as e: 

73 raise IPCFlagException("failed to create %s: %s" % (flagFile, e)) 

74 

75 def test(self, name): 

76 """Test the flag""" 

77 flagFile = os.path.join(self.nsDir, name) 

78 return util.pathexists(flagFile) 

79 

80 def clear(self, name): 

81 """Clear the flag""" 

82 if self.test(name): 

83 flagFile = os.path.join(self.nsDir, name) 

84 try: 

85 os.unlink(flagFile) 

86 util.SMlog("IPCFlag: clear %s:%s" % (self.ns, name)) 

87 except OSError: 

88 raise IPCFlagException("failed to remove %s" % flagFile) 

89 

90 def clearAll(self): 

91 try: 

92 for file in os.listdir(self.nsDir): 

93 path = os.path.join(self.nsDir, file) 

94 os.unlink(path) 

95 except OSError: 

96 raise IPCFlagException("failed to remove %s" % path) 

97 

98 

99def _runTests(): 

100 flag = IPCFlag("A") 

101 flag.set("X") 

102 assert flag.test("X") 

103 flag.clear("X") 

104 assert not flag.test("X") 

105 assert not flag.test("Y") 

106 flag.set("X") 

107 flag.set("Y") 

108 flag.set("Z") 

109 assert flag.test("X") 

110 assert flag.test("Y") 

111 assert flag.test("Z") 

112 flag.clearAll() 

113 assert not flag.test("X") 

114 assert not flag.test("Y") 

115 assert not flag.test("Z") 

116 print("All tests passed") 

117 

118if __name__ == '__main__': 118 ↛ 119line 118 didn't jump to line 119, because the condition on line 118 was never true

119 _runTests()