Skip to content

concurrent

Classes:

ConcurrentStore 

ConcurrentStore(store)

A store that allows concurrent reads and writes.

Methods:

Attributes:

Source code in rdflib/plugins/stores/concurrent.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, store):
    self.store = store

    # number of calls to visit still in progress
    self.__visit_count = 0

    # lock for locking down the indices
    self.__lock = Lock()

    # lists for keeping track of added and removed triples while
    # we wait for the lock
    self.__pending_removes = []
    self.__pending_adds = []

__lock instance-attribute 

__lock = Lock()

__pending_adds instance-attribute 

__pending_adds = []

__pending_removes instance-attribute 

__pending_removes = []

__visit_count instance-attribute 

__visit_count = 0

store instance-attribute 

store = store

__begin_read 

__begin_read()
Source code in rdflib/plugins/stores/concurrent.py 
74
75
76
77
78
def __begin_read(self):
    lock = self.__lock
    lock.acquire()
    self.__visit_count = self.__visit_count + 1
    lock.release()

__end_read 

__end_read()
Source code in rdflib/plugins/stores/concurrent.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __end_read(self):
    lock = self.__lock
    lock.acquire()
    self.__visit_count = self.__visit_count - 1
    if self.__visit_count == 0:
        pending_removes = self.__pending_removes
        while pending_removes:
            (s, p, o) = pending_removes.pop()
            try:
                self.store.remove((s, p, o))
            except:  # noqa: E722
                # TODO: change to try finally?
                print(s, p, o, "Not in store to remove")
        pending_adds = self.__pending_adds
        while pending_adds:
            (s, p, o) = pending_adds.pop()
            self.store.add((s, p, o))
    lock.release()

__len__ 

__len__()
Source code in rdflib/plugins/stores/concurrent.py 
71
72
def __len__(self):
    return self.store.__len__()

add 

add(triple)
Source code in rdflib/plugins/stores/concurrent.py 
40
41
42
43
44
45
def add(self, triple):
    (s, p, o) = triple
    if self.__visit_count == 0:
        self.store.add((s, p, o))
    else:
        self.__pending_adds.append((s, p, o))

remove 

remove(triple)
Source code in rdflib/plugins/stores/concurrent.py 
47
48
49
50
51
52
def remove(self, triple):
    (s, p, o) = triple
    if self.__visit_count == 0:
        self.store.remove((s, p, o))
    else:
        self.__pending_removes.append((s, p, o))

triples 

triples(triple)
Source code in rdflib/plugins/stores/concurrent.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def triples(self, triple):
    (su, pr, ob) = triple
    g = self.store.triples((su, pr, ob))
    pending_removes = self.__pending_removes
    self.__begin_read()
    for s, p, o in ResponsibleGenerator(g, self.__end_read):
        if not (s, p, o) in pending_removes:  # noqa: E713
            yield s, p, o

    for s, p, o in self.__pending_adds:
        if (
            (su is None or su == s)
            and (pr is None or pr == p)
            and (ob is None or ob == o)
        ):
            yield s, p, o

ResponsibleGenerator 

ResponsibleGenerator(gen, cleanup)

A generator that will help clean up when it is done being used.

Methods:

Attributes:

Source code in rdflib/plugins/stores/concurrent.py 
 9
10
11
def __init__(self, gen, cleanup):
    self.cleanup = cleanup
    self.gen = gen

__slots__ class-attribute instance-attribute 

__slots__ = ['cleanup', 'gen']

cleanup instance-attribute 

cleanup = cleanup

gen instance-attribute 

gen = gen

__del__ 

__del__()
Source code in rdflib/plugins/stores/concurrent.py 
13
14
def __del__(self):
    self.cleanup()

__iter__ 

__iter__()
Source code in rdflib/plugins/stores/concurrent.py 
16
17
def __iter__(self):
    return self

__next__ 

__next__()
Source code in rdflib/plugins/stores/concurrent.py 
19
20
def __next__(self):
    return next(self.gen)