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

#!/usr/bin/python 

# 

# Copyright (C) Citrix Systems Inc. 

# 

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

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

# by the Free Software Foundation; version 2.1 only. 

# 

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

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

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the  

# GNU Lesser General Public License for more details. 

# 

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

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

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

 

"""Communication for processes""" 

 

import os 

import util 

import errno 

 

class IPCFlagException(util.SMException): 

    pass 

 

class IPCFlag: 

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

    Not thread-safe.""" 

 

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

 

    def __init__(self, ns): 

        self.ns = ns 

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

        if not util.pathexists(self.nsDir): 

            try: 

                os.makedirs(self.nsDir) 

            except OSError: 

                pass 

        if not util.pathexists(self.nsDir): 

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

 

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

        """Set the flag 

 

        name: the file to set 

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

        trying to set create, the file MAY be overwritten. 

 

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

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

            return 

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

        try: 

            if soft: 

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

            else: 

                f = open(flagFile, "w") 

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

            f.close() 

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

            return True 

        except OSError, e: 

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

                return False 

            else: 

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

                        % (flagFile, e)) 

        except IOError, e: 

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

 

    def test(self, name): 

        """Test the flag""" 

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

        return util.pathexists(flagFile) 

 

    def clear(self, name): 

        """Clear the flag""" 

        if self.test(name): 

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

            try: 

                os.unlink(flagFile) 

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

            except OSError: 

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

 

    def clearAll(self): 

        try: 

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

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

                os.unlink(path) 

        except OSError: 

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

 

def _runTests(): 

    flag = IPCFlag("A") 

    flag.set("X") 

    assert flag.test("X") 

    flag.clear("X") 

    assert not flag.test("X") 

    assert not flag.test("Y") 

    flag.set("X") 

    flag.set("Y") 

    flag.set("Z") 

    assert flag.test("X") 

    assert flag.test("Y") 

    assert flag.test("Z") 

    flag.clearAll() 

    assert not flag.test("X") 

    assert not flag.test("Y") 

    assert not flag.test("Z") 

    print "All tests passed" 

 

116if __name__ == '__main__': 

    _runTests()