We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Allow self.master to be set so it can be used before service setup.
This also addresses a few testing issues, caches the master property,
and invalidates the cache on disownServiceParent.  Fixes #3329.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `master/buildbot/db/schedulers.py`
Content:
```
1 # This file is part of Buildbot.  Buildbot is free software: you can
2 # redistribute it and/or modify it under the terms of the GNU General Public
3 # License as published by the Free Software Foundation, version 2.
4 #
5 # This program is distributed in the hope that it will be useful, but WITHOUT
6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
8 # details.
9 #
10 # You should have received a copy of the GNU General Public License along with
11 # this program; if not, write to the Free Software Foundation, Inc., 51
12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
13 #
14 # Copyright Buildbot Team Members
15 
16 import sqlalchemy as sa
17 import sqlalchemy.exc
18 from buildbot.db import base
19 
20 class SchedulersConnectorComponent(base.DBConnectorComponent):
21     # Documentation is in developer/database.rst
22 
23     def classifyChanges(self, objectid, classifications):
24         def thd(conn):
25             transaction = conn.begin()
26             tbl = self.db.model.scheduler_changes
27             ins_q = tbl.insert()
28             upd_q = tbl.update(
29                     ((tbl.c.objectid == objectid)
30                     & (tbl.c.changeid == sa.bindparam('wc_changeid'))))
31             for changeid, important in classifications.items():
32                 # convert the 'important' value into an integer, since that
33                 # is the column type
34                 imp_int = important and 1 or 0
35                 try:
36                     conn.execute(ins_q,
37                             objectid=objectid,
38                             changeid=changeid,
39                             important=imp_int)
40                 except (sqlalchemy.exc.ProgrammingError,
41                         sqlalchemy.exc.IntegrityError):
42                     # insert failed, so try an update
43                     conn.execute(upd_q,
44                             wc_changeid=changeid,
45                             important=imp_int)
46 
47             transaction.commit()
48         return self.db.pool.do(thd)
49 
50     def flushChangeClassifications(self, objectid, less_than=None):
51         def thd(conn):
52             sch_ch_tbl = self.db.model.scheduler_changes
53             wc = (sch_ch_tbl.c.objectid == objectid)
54             if less_than is not None:
55                 wc = wc & (sch_ch_tbl.c.changeid < less_than)
56             q = sch_ch_tbl.delete(whereclause=wc)
57             conn.execute(q)
58         return self.db.pool.do(thd)
59 
60     class Thunk: pass
61     def getChangeClassifications(self, objectid, branch=Thunk):
62         def thd(conn):
63             sch_ch_tbl = self.db.model.scheduler_changes
64             ch_tbl = self.db.model.changes
65 
66             wc = (sch_ch_tbl.c.objectid == objectid)
67             if branch is not self.Thunk:
68                 wc = wc & (
69                     (sch_ch_tbl.c.changeid == ch_tbl.c.changeid) &
70                     (ch_tbl.c.branch == branch))
71             q = sa.select(
72                 [ sch_ch_tbl.c.changeid, sch_ch_tbl.c.important ],
73                 whereclause=wc)
74             return dict([ (r.changeid, [False,True][r.important])
75                           for r in conn.execute(q) ])
76         return self.db.pool.do(thd)
77
```

Path: `master/buildbot/schedulers/basic.py`
Content:
```
1 # This file is part of Buildbot.  Buildbot is free software: you can
2 # redistribute it and/or modify it under the terms of the GNU General Public
3 # License as published by the Free Software Foundation, version 2.
4 #
5 # This program is distributed in the hope that it will be useful, but WITHOUT
6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
8 # details.
9 #
10 # You should have received a copy of the GNU General Public License along with
11 # this program; if not, write to the Free Software Foundation, Inc., 51
12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
13 #
14 # Copyright Buildbot Team Members
15 
16 from twisted.internet import defer, reactor
17 from twisted.python import log
18 from buildbot import util, config
19 from buildbot.util import NotABranch
20 from collections import defaultdict
21 from buildbot.changes import filter, changes
22 from buildbot.schedulers import base, dependent
23 
24 class BaseBasicScheduler(base.BaseScheduler):
25     """
26     @param onlyImportant: If True, only important changes will be added to the
27                           buildset.
28     @type onlyImportant: boolean
29 
30     """
31 
32     compare_attrs = (base.BaseScheduler.compare_attrs +
33                      ('treeStableTimer', 'change_filter', 'fileIsImportant',
34                       'onlyImportant') )
35 
36     _reactor = reactor # for tests
37 
38     fileIsImportant = None
39 
40     class NotSet: pass
41     def __init__(self, name, shouldntBeSet=NotSet, treeStableTimer=None,
42                 builderNames=None, branch=NotABranch, branches=NotABranch,
43                 fileIsImportant=None, properties={}, categories=None,
44                 change_filter=None, onlyImportant=False, **kwargs):
45         if shouldntBeSet is not self.NotSet:
46             config.error(
47                 "pass arguments to schedulers using keyword arguments")
48         if fileIsImportant and not callable(fileIsImportant):
49             config.error(
50                 "fileIsImportant must be a callable")
51 
52         # initialize parent classes
53         base.BaseScheduler.__init__(self, name, builderNames, properties, **kwargs)
54 
55         self.treeStableTimer = treeStableTimer
56         if fileIsImportant is not None:
57             self.fileIsImportant = fileIsImportant
58         self.onlyImportant = onlyImportant
59         self.change_filter = self.getChangeFilter(branch=branch,
60                 branches=branches, change_filter=change_filter,
61                 categories=categories)
62 
63         # the IDelayedCall used to wake up when this scheduler's
64         # treeStableTimer expires.
65         self._stable_timers = defaultdict(lambda : None)
66         self._stable_timers_lock = defer.DeferredLock()
67 
68     def getChangeFilter(self, branch, branches, change_filter, categories):
69         raise NotImplementedError
70 
71     def startService(self, _returnDeferred=False):
72         base.BaseScheduler.startService(self)
73 
74         d = self.startConsumingChanges(fileIsImportant=self.fileIsImportant,
75                                        change_filter=self.change_filter,
76                                        onlyImportant=self.onlyImportant)
77 
78         # if treeStableTimer is False, then we don't care about classified
79         # changes, so get rid of any hanging around from previous
80         # configurations
81         if not self.treeStableTimer:
82             d.addCallback(lambda _ :
83                 self.master.db.schedulers.flushChangeClassifications(
84                                                         self.objectid))
85 
86         # otherwise, if there are classified changes out there, start their
87         # treeStableTimers again
88         else:
89             d.addCallback(lambda _ :
90                 self.scanExistingClassifiedChanges())
91 
92         # handle Deferred errors, since startService does not return a Deferred
93         d.addErrback(log.err, "while starting SingleBranchScheduler '%s'"
94                               % self.name)
95 
96         if _returnDeferred:
97             return d # only used in tests
98 
99     def stopService(self):
100         # the base stopService will unsubscribe from new changes
101         d = base.BaseScheduler.stopService(self)
102         @util.deferredLocked(self._stable_timers_lock)
103         def cancel_timers(_):
104             for timer in self._stable_timers.values():
105                 if timer:
106                     timer.cancel()
107             self._stable_timers.clear()
108         d.addCallback(cancel_timers)
109         return d
110 
111     @util.deferredLocked('_stable_timers_lock')
112     def gotChange(self, change, important):
113         if not self.treeStableTimer:
114             # if there's no treeStableTimer, we can completely ignore
115             # unimportant changes
116             if not important:
117                 return defer.succeed(None)
118             # otherwise, we'll build it right away
119             return self.addBuildsetForChanges(reason='scheduler',
120                             changeids=[ change.number ])
121 
122         timer_name = self.getTimerNameForChange(change)
123 
124         # if we have a treeStableTimer, then record the change's importance
125         # and:
126         # - for an important change, start the timer
127         # - for an unimportant change, reset the timer if it is running
128         d = self.master.db.schedulers.classifyChanges(
129                 self.objectid, { change.number : important })
130         def fix_timer(_):
131             if not important and not self._stable_timers[timer_name]:
132                 return
133             if self._stable_timers[timer_name]:
134                 self._stable_timers[timer_name].cancel()
135             def fire_timer():
136                 d = self.stableTimerFired(timer_name)
137                 d.addErrback(log.err, "while firing stable timer")
138             self._stable_timers[timer_name] = self._reactor.callLater(
139                     self.treeStableTimer, fire_timer)
140         d.addCallback(fix_timer)
141         return d
142 
143     @defer.inlineCallbacks
144     def scanExistingClassifiedChanges(self):
145         # call gotChange for each classified change.  This is called at startup
146         # and is intended to re-start the treeStableTimer for any changes that
147         # had not yet been built when the scheduler was stopped.
148 
149         # NOTE: this may double-call gotChange for changes that arrive just as
150         # the scheduler starts up.  In practice, this doesn't hurt anything.
151         classifications = \
152                 yield self.master.db.schedulers.getChangeClassifications(
153                                                                 self.objectid)
154 
155         # call gotChange for each change, after first fetching it from the db
156         for changeid, important in classifications.iteritems():
157             chdict = yield self.master.db.changes.getChange(changeid)
158 
159             if not chdict:
160                 continue
161 
162             change = yield changes.Change.fromChdict(self.master, chdict)
163             yield self.gotChange(change, important)
164 
165     def getTimerNameForChange(self, change):
166         raise NotImplementedError # see subclasses
167 
168     def getChangeClassificationsForTimer(self, objectid, timer_name):
169         """similar to db.schedulers.getChangeClassifications, but given timer
170         name"""
171         raise NotImplementedError # see subclasses
172 
173     @util.deferredLocked('_stable_timers_lock')
174     @defer.inlineCallbacks
175     def stableTimerFired(self, timer_name):
176         # if the service has already been stoppd then just bail out
177         if not self._stable_timers[timer_name]:
178             return
179 
180         # delete this now-fired timer
181         del self._stable_timers[timer_name]
182 
183         classifications = \
184             yield self.getChangeClassificationsForTimer(self.objectid,
185                                                             timer_name)
186 
187         # just in case: databases do weird things sometimes!
188         if not classifications: # pragma: no cover
189             return
190 
191         changeids = sorted(classifications.keys())
192         yield self.addBuildsetForChanges(reason='scheduler',
193                                            changeids=changeids)
194 
195         max_changeid = changeids[-1] # (changeids are sorted)
196         yield self.master.db.schedulers.flushChangeClassifications(
197                             self.objectid, less_than=max_changeid+1)
198 
199     def getPendingBuildTimes(self):
200         # This isn't locked, since the caller expects and immediate value,
201         # and in any case, this is only an estimate.
202         return [timer.getTime() for timer in self._stable_timers.values() if timer and timer.active()]
203 
204 class SingleBranchScheduler(BaseBasicScheduler):
205     def getChangeFilter(self, branch, branches, change_filter, categories):
206         if branch is NotABranch and not change_filter:
207             config.error(
208                 "the 'branch' argument to SingleBranchScheduler is " +
209                 "mandatory unless change_filter is provided")
210         elif branches is not NotABranch:
211             config.error(
212                 "the 'branches' argument is not allowed for " +
213                 "SingleBranchScheduler")
214 
215 
216         return filter.ChangeFilter.fromSchedulerConstructorArgs(
217                 change_filter=change_filter, branch=branch,
218                 categories=categories)
219 
220     def getTimerNameForChange(self, change):
221         return "only" # this class only uses one timer
222 
223     def getChangeClassificationsForTimer(self, objectid, timer_name):
224         return self.master.db.schedulers.getChangeClassifications(
225                                                         self.objectid)
226 
227 
228 class Scheduler(SingleBranchScheduler):
229     "alias for SingleBranchScheduler"
230     def __init__(self, *args, **kwargs):
231         log.msg("WARNING: the name 'Scheduler' is deprecated; use " +
232                 "buildbot.schedulers.basic.SingleBranchScheduler instead " +
233                 "(note that this may require you to change your import " +
234                 "statement)")
235         SingleBranchScheduler.__init__(self, *args, **kwargs)
236 
237 
238 class AnyBranchScheduler(BaseBasicScheduler):
239     def getChangeFilter(self, branch, branches, change_filter, categories):
240         assert branch is NotABranch
241         return filter.ChangeFilter.fromSchedulerConstructorArgs(
242                 change_filter=change_filter, branch=branches,
243                 categories=categories)
244 
245     def getTimerNameForChange(self, change):
246         return change.branch
247 
248     def getChangeClassificationsForTimer(self, objectid, timer_name):
249         branch = timer_name # set in getTimerNameForChange
250         return self.master.db.schedulers.getChangeClassifications(
251                 self.objectid, branch=branch)
252 
253 # now at buildbot.schedulers.dependent, but keep the old name alive
254 Dependent = dependent.Dependent
255
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/master/buildbot/db/schedulers.py b/master/buildbot/db/schedulers.py
--- a/master/buildbot/db/schedulers.py
+++ b/master/buildbot/db/schedulers.py
@@ -58,16 +58,33 @@
         return self.db.pool.do(thd)
 
     class Thunk: pass
-    def getChangeClassifications(self, objectid, branch=Thunk):
+    def getChangeClassifications(self, objectid, branch=Thunk,
+                                 repository=Thunk, project=Thunk,
+                                 codebase=Thunk):
         def thd(conn):
             sch_ch_tbl = self.db.model.scheduler_changes
             ch_tbl = self.db.model.changes
 
             wc = (sch_ch_tbl.c.objectid == objectid)
+                        
+            # may need to filter further based on branch, etc
+            extra_wheres = [] 
             if branch is not self.Thunk:
-                wc = wc & (
-                    (sch_ch_tbl.c.changeid == ch_tbl.c.changeid) &
-                    (ch_tbl.c.branch == branch))
+                extra_wheres.append(ch_tbl.c.branch == branch)
+            if repository is not self.Thunk:
+                extra_wheres.append(ch_tbl.c.repository == repository)
+            if project is not self.Thunk:
+                extra_wheres.append(ch_tbl.c.project == project)
+            if codebase is not self.Thunk:
+                extra_wheres.append(ch_tbl.c.codebase == codebase)
+
+            # if we need to filter further append those, as well as a join
+            # on changeid (but just once for that one)
+            if extra_wheres:
+                wc &= (sch_ch_tbl.c.changeid == ch_tbl.c.changeid)
+                for w in extra_wheres:
+                    wc &= w
+
             q = sa.select(
                 [ sch_ch_tbl.c.changeid, sch_ch_tbl.c.important ],
                 whereclause=wc)
diff --git a/master/buildbot/schedulers/basic.py b/master/buildbot/schedulers/basic.py
--- a/master/buildbot/schedulers/basic.py
+++ b/master/buildbot/schedulers/basic.py
@@ -243,12 +243,14 @@
                 categories=categories)
 
     def getTimerNameForChange(self, change):
-        return change.branch
+        # Py2.6+: could be a namedtuple
+        return (change.codebase, change.project, change.repository, change.branch)
 
     def getChangeClassificationsForTimer(self, objectid, timer_name):
-        branch = timer_name # set in getTimerNameForChange
+        codebase, project, repository, branch = timer_name # set in getTimerNameForChange
         return self.master.db.schedulers.getChangeClassifications(
-                self.objectid, branch=branch)
+                self.objectid, branch=branch, repository=repository,
+                codebase=codebase, project=project)
 
 # now at buildbot.schedulers.dependent, but keep the old name alive
 Dependent = dependent.Dependent

{"golden_diff": "diff --git a/master/buildbot/db/schedulers.py b/master/buildbot/db/schedulers.py\n--- a/master/buildbot/db/schedulers.py\n+++ b/master/buildbot/db/schedulers.py\n@@ -58,16 +58,33 @@\n         return self.db.pool.do(thd)\n \n     class Thunk: pass\n-    def getChangeClassifications(self, objectid, branch=Thunk):\n+    def getChangeClassifications(self, objectid, branch=Thunk,\n+                                 repository=Thunk, project=Thunk,\n+                                 codebase=Thunk):\n         def thd(conn):\n             sch_ch_tbl = self.db.model.scheduler_changes\n             ch_tbl = self.db.model.changes\n \n             wc = (sch_ch_tbl.c.objectid == objectid)\n+                        \n+            # may need to filter further based on branch, etc\n+            extra_wheres = [] \n             if branch is not self.Thunk:\n-                wc = wc & (\n-                    (sch_ch_tbl.c.changeid == ch_tbl.c.changeid) &\n-                    (ch_tbl.c.branch == branch))\n+                extra_wheres.append(ch_tbl.c.branch == branch)\n+            if repository is not self.Thunk:\n+                extra_wheres.append(ch_tbl.c.repository == repository)\n+            if project is not self.Thunk:\n+                extra_wheres.append(ch_tbl.c.project == project)\n+            if codebase is not self.Thunk:\n+                extra_wheres.append(ch_tbl.c.codebase == codebase)\n+\n+            # if we need to filter further append those, as well as a join\n+            # on changeid (but just once for that one)\n+            if extra_wheres:\n+                wc &= (sch_ch_tbl.c.changeid == ch_tbl.c.changeid)\n+                for w in extra_wheres:\n+                    wc &= w\n+\n             q = sa.select(\n                 [ sch_ch_tbl.c.changeid, sch_ch_tbl.c.important ],\n                 whereclause=wc)\ndiff --git a/master/buildbot/schedulers/basic.py b/master/buildbot/schedulers/basic.py\n--- a/master/buildbot/schedulers/basic.py\n+++ b/master/buildbot/schedulers/basic.py\n@@ -243,12 +243,14 @@\n                 categories=categories)\n \n     def getTimerNameForChange(self, change):\n-        return change.branch\n+        # Py2.6+: could be a namedtuple\n+        return (change.codebase, change.project, change.repository, change.branch)\n \n     def getChangeClassificationsForTimer(self, objectid, timer_name):\n-        branch = timer_name # set in getTimerNameForChange\n+        codebase, project, repository, branch = timer_name # set in getTimerNameForChange\n         return self.master.db.schedulers.getChangeClassifications(\n-                self.objectid, branch=branch)\n+                self.objectid, branch=branch, repository=repository,\n+                codebase=codebase, project=project)\n \n # now at buildbot.schedulers.dependent, but keep the old name alive\n Dependent = dependent.Dependent\n", "issue": "Allow self.master to be set so it can be used before service setup.\nThis also addresses a few testing issues, caches the master property,\nand invalidates the cache on disownServiceParent.  Fixes #3329.\n\n", "before_files": [{"content": "# This file is part of Buildbot.  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright Buildbot Team Members\n\nimport sqlalchemy as sa\nimport sqlalchemy.exc\nfrom buildbot.db import base\n\nclass SchedulersConnectorComponent(base.DBConnectorComponent):\n    # Documentation is in developer/database.rst\n\n    def classifyChanges(self, objectid, classifications):\n        def thd(conn):\n            transaction = conn.begin()\n            tbl = self.db.model.scheduler_changes\n            ins_q = tbl.insert()\n            upd_q = tbl.update(\n                    ((tbl.c.objectid == objectid)\n                    & (tbl.c.changeid == sa.bindparam('wc_changeid'))))\n            for changeid, important in classifications.items():\n                # convert the 'important' value into an integer, since that\n                # is the column type\n                imp_int = important and 1 or 0\n                try:\n                    conn.execute(ins_q,\n                            objectid=objectid,\n                            changeid=changeid,\n                            important=imp_int)\n                except (sqlalchemy.exc.ProgrammingError,\n                        sqlalchemy.exc.IntegrityError):\n                    # insert failed, so try an update\n                    conn.execute(upd_q,\n                            wc_changeid=changeid,\n                            important=imp_int)\n\n            transaction.commit()\n        return self.db.pool.do(thd)\n\n    def flushChangeClassifications(self, objectid, less_than=None):\n        def thd(conn):\n            sch_ch_tbl = self.db.model.scheduler_changes\n            wc = (sch_ch_tbl.c.objectid == objectid)\n            if less_than is not None:\n                wc = wc & (sch_ch_tbl.c.changeid < less_than)\n            q = sch_ch_tbl.delete(whereclause=wc)\n            conn.execute(q)\n        return self.db.pool.do(thd)\n\n    class Thunk: pass\n    def getChangeClassifications(self, objectid, branch=Thunk):\n        def thd(conn):\n            sch_ch_tbl = self.db.model.scheduler_changes\n            ch_tbl = self.db.model.changes\n\n            wc = (sch_ch_tbl.c.objectid == objectid)\n            if branch is not self.Thunk:\n                wc = wc & (\n                    (sch_ch_tbl.c.changeid == ch_tbl.c.changeid) &\n                    (ch_tbl.c.branch == branch))\n            q = sa.select(\n                [ sch_ch_tbl.c.changeid, sch_ch_tbl.c.important ],\n                whereclause=wc)\n            return dict([ (r.changeid, [False,True][r.important])\n                          for r in conn.execute(q) ])\n        return self.db.pool.do(thd)\n", "path": "master/buildbot/db/schedulers.py"}, {"content": "# This file is part of Buildbot.  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright Buildbot Team Members\n\nfrom twisted.internet import defer, reactor\nfrom twisted.python import log\nfrom buildbot import util, config\nfrom buildbot.util import NotABranch\nfrom collections import defaultdict\nfrom buildbot.changes import filter, changes\nfrom buildbot.schedulers import base, dependent\n\nclass BaseBasicScheduler(base.BaseScheduler):\n    \"\"\"\n    @param onlyImportant: If True, only important changes will be added to the\n                          buildset.\n    @type onlyImportant: boolean\n\n    \"\"\"\n\n    compare_attrs = (base.BaseScheduler.compare_attrs +\n                     ('treeStableTimer', 'change_filter', 'fileIsImportant',\n                      'onlyImportant') )\n\n    _reactor = reactor # for tests\n\n    fileIsImportant = None\n\n    class NotSet: pass\n    def __init__(self, name, shouldntBeSet=NotSet, treeStableTimer=None,\n                builderNames=None, branch=NotABranch, branches=NotABranch,\n                fileIsImportant=None, properties={}, categories=None,\n                change_filter=None, onlyImportant=False, **kwargs):\n        if shouldntBeSet is not self.NotSet:\n            config.error(\n                \"pass arguments to schedulers using keyword arguments\")\n        if fileIsImportant and not callable(fileIsImportant):\n            config.error(\n                \"fileIsImportant must be a callable\")\n\n        # initialize parent classes\n        base.BaseScheduler.__init__(self, name, builderNames, properties, **kwargs)\n\n        self.treeStableTimer = treeStableTimer\n        if fileIsImportant is not None:\n            self.fileIsImportant = fileIsImportant\n        self.onlyImportant = onlyImportant\n        self.change_filter = self.getChangeFilter(branch=branch,\n                branches=branches, change_filter=change_filter,\n                categories=categories)\n\n        # the IDelayedCall used to wake up when this scheduler's\n        # treeStableTimer expires.\n        self._stable_timers = defaultdict(lambda : None)\n        self._stable_timers_lock = defer.DeferredLock()\n\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        raise NotImplementedError\n\n    def startService(self, _returnDeferred=False):\n        base.BaseScheduler.startService(self)\n\n        d = self.startConsumingChanges(fileIsImportant=self.fileIsImportant,\n                                       change_filter=self.change_filter,\n                                       onlyImportant=self.onlyImportant)\n\n        # if treeStableTimer is False, then we don't care about classified\n        # changes, so get rid of any hanging around from previous\n        # configurations\n        if not self.treeStableTimer:\n            d.addCallback(lambda _ :\n                self.master.db.schedulers.flushChangeClassifications(\n                                                        self.objectid))\n\n        # otherwise, if there are classified changes out there, start their\n        # treeStableTimers again\n        else:\n            d.addCallback(lambda _ :\n                self.scanExistingClassifiedChanges())\n\n        # handle Deferred errors, since startService does not return a Deferred\n        d.addErrback(log.err, \"while starting SingleBranchScheduler '%s'\"\n                              % self.name)\n\n        if _returnDeferred:\n            return d # only used in tests\n\n    def stopService(self):\n        # the base stopService will unsubscribe from new changes\n        d = base.BaseScheduler.stopService(self)\n        @util.deferredLocked(self._stable_timers_lock)\n        def cancel_timers(_):\n            for timer in self._stable_timers.values():\n                if timer:\n                    timer.cancel()\n            self._stable_timers.clear()\n        d.addCallback(cancel_timers)\n        return d\n\n    @util.deferredLocked('_stable_timers_lock')\n    def gotChange(self, change, important):\n        if not self.treeStableTimer:\n            # if there's no treeStableTimer, we can completely ignore\n            # unimportant changes\n            if not important:\n                return defer.succeed(None)\n            # otherwise, we'll build it right away\n            return self.addBuildsetForChanges(reason='scheduler',\n                            changeids=[ change.number ])\n\n        timer_name = self.getTimerNameForChange(change)\n\n        # if we have a treeStableTimer, then record the change's importance\n        # and:\n        # - for an important change, start the timer\n        # - for an unimportant change, reset the timer if it is running\n        d = self.master.db.schedulers.classifyChanges(\n                self.objectid, { change.number : important })\n        def fix_timer(_):\n            if not important and not self._stable_timers[timer_name]:\n                return\n            if self._stable_timers[timer_name]:\n                self._stable_timers[timer_name].cancel()\n            def fire_timer():\n                d = self.stableTimerFired(timer_name)\n                d.addErrback(log.err, \"while firing stable timer\")\n            self._stable_timers[timer_name] = self._reactor.callLater(\n                    self.treeStableTimer, fire_timer)\n        d.addCallback(fix_timer)\n        return d\n\n    @defer.inlineCallbacks\n    def scanExistingClassifiedChanges(self):\n        # call gotChange for each classified change.  This is called at startup\n        # and is intended to re-start the treeStableTimer for any changes that\n        # had not yet been built when the scheduler was stopped.\n\n        # NOTE: this may double-call gotChange for changes that arrive just as\n        # the scheduler starts up.  In practice, this doesn't hurt anything.\n        classifications = \\\n                yield self.master.db.schedulers.getChangeClassifications(\n                                                                self.objectid)\n\n        # call gotChange for each change, after first fetching it from the db\n        for changeid, important in classifications.iteritems():\n            chdict = yield self.master.db.changes.getChange(changeid)\n\n            if not chdict:\n                continue\n\n            change = yield changes.Change.fromChdict(self.master, chdict)\n            yield self.gotChange(change, important)\n\n    def getTimerNameForChange(self, change):\n        raise NotImplementedError # see subclasses\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        \"\"\"similar to db.schedulers.getChangeClassifications, but given timer\n        name\"\"\"\n        raise NotImplementedError # see subclasses\n\n    @util.deferredLocked('_stable_timers_lock')\n    @defer.inlineCallbacks\n    def stableTimerFired(self, timer_name):\n        # if the service has already been stoppd then just bail out\n        if not self._stable_timers[timer_name]:\n            return\n\n        # delete this now-fired timer\n        del self._stable_timers[timer_name]\n\n        classifications = \\\n            yield self.getChangeClassificationsForTimer(self.objectid,\n                                                            timer_name)\n\n        # just in case: databases do weird things sometimes!\n        if not classifications: # pragma: no cover\n            return\n\n        changeids = sorted(classifications.keys())\n        yield self.addBuildsetForChanges(reason='scheduler',\n                                           changeids=changeids)\n\n        max_changeid = changeids[-1] # (changeids are sorted)\n        yield self.master.db.schedulers.flushChangeClassifications(\n                            self.objectid, less_than=max_changeid+1)\n\n    def getPendingBuildTimes(self):\n        # This isn't locked, since the caller expects and immediate value,\n        # and in any case, this is only an estimate.\n        return [timer.getTime() for timer in self._stable_timers.values() if timer and timer.active()]\n\nclass SingleBranchScheduler(BaseBasicScheduler):\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        if branch is NotABranch and not change_filter:\n            config.error(\n                \"the 'branch' argument to SingleBranchScheduler is \" +\n                \"mandatory unless change_filter is provided\")\n        elif branches is not NotABranch:\n            config.error(\n                \"the 'branches' argument is not allowed for \" +\n                \"SingleBranchScheduler\")\n\n\n        return filter.ChangeFilter.fromSchedulerConstructorArgs(\n                change_filter=change_filter, branch=branch,\n                categories=categories)\n\n    def getTimerNameForChange(self, change):\n        return \"only\" # this class only uses one timer\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        return self.master.db.schedulers.getChangeClassifications(\n                                                        self.objectid)\n\n\nclass Scheduler(SingleBranchScheduler):\n    \"alias for SingleBranchScheduler\"\n    def __init__(self, *args, **kwargs):\n        log.msg(\"WARNING: the name 'Scheduler' is deprecated; use \" +\n                \"buildbot.schedulers.basic.SingleBranchScheduler instead \" +\n                \"(note that this may require you to change your import \" +\n                \"statement)\")\n        SingleBranchScheduler.__init__(self, *args, **kwargs)\n\n\nclass AnyBranchScheduler(BaseBasicScheduler):\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        assert branch is NotABranch\n        return filter.ChangeFilter.fromSchedulerConstructorArgs(\n                change_filter=change_filter, branch=branches,\n                categories=categories)\n\n    def getTimerNameForChange(self, change):\n        return change.branch\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        branch = timer_name # set in getTimerNameForChange\n        return self.master.db.schedulers.getChangeClassifications(\n                self.objectid, branch=branch)\n\n# now at buildbot.schedulers.dependent, but keep the old name alive\nDependent = dependent.Dependent\n", "path": "master/buildbot/schedulers/basic.py"}], "after_files": [{"content": "# This file is part of Buildbot.  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright Buildbot Team Members\n\nimport sqlalchemy as sa\nimport sqlalchemy.exc\nfrom buildbot.db import base\n\nclass SchedulersConnectorComponent(base.DBConnectorComponent):\n    # Documentation is in developer/database.rst\n\n    def classifyChanges(self, objectid, classifications):\n        def thd(conn):\n            transaction = conn.begin()\n            tbl = self.db.model.scheduler_changes\n            ins_q = tbl.insert()\n            upd_q = tbl.update(\n                    ((tbl.c.objectid == objectid)\n                    & (tbl.c.changeid == sa.bindparam('wc_changeid'))))\n            for changeid, important in classifications.items():\n                # convert the 'important' value into an integer, since that\n                # is the column type\n                imp_int = important and 1 or 0\n                try:\n                    conn.execute(ins_q,\n                            objectid=objectid,\n                            changeid=changeid,\n                            important=imp_int)\n                except (sqlalchemy.exc.ProgrammingError,\n                        sqlalchemy.exc.IntegrityError):\n                    # insert failed, so try an update\n                    conn.execute(upd_q,\n                            wc_changeid=changeid,\n                            important=imp_int)\n\n            transaction.commit()\n        return self.db.pool.do(thd)\n\n    def flushChangeClassifications(self, objectid, less_than=None):\n        def thd(conn):\n            sch_ch_tbl = self.db.model.scheduler_changes\n            wc = (sch_ch_tbl.c.objectid == objectid)\n            if less_than is not None:\n                wc = wc & (sch_ch_tbl.c.changeid < less_than)\n            q = sch_ch_tbl.delete(whereclause=wc)\n            conn.execute(q)\n        return self.db.pool.do(thd)\n\n    class Thunk: pass\n    def getChangeClassifications(self, objectid, branch=Thunk,\n                                 repository=Thunk, project=Thunk,\n                                 codebase=Thunk):\n        def thd(conn):\n            sch_ch_tbl = self.db.model.scheduler_changes\n            ch_tbl = self.db.model.changes\n\n            wc = (sch_ch_tbl.c.objectid == objectid)\n                        \n            # may need to filter further based on branch, etc\n            extra_wheres = [] \n            if branch is not self.Thunk:\n                extra_wheres.append(ch_tbl.c.branch == branch)\n            if repository is not self.Thunk:\n                extra_wheres.append(ch_tbl.c.repository == repository)\n            if project is not self.Thunk:\n                extra_wheres.append(ch_tbl.c.project == project)\n            if codebase is not self.Thunk:\n                extra_wheres.append(ch_tbl.c.codebase == codebase)\n\n            # if we need to filter further append those, as well as a join\n            # on changeid (but just once for that one)\n            if extra_wheres:\n                wc &= (sch_ch_tbl.c.changeid == ch_tbl.c.changeid)\n                for w in extra_wheres:\n                    wc &= w\n\n            q = sa.select(\n                [ sch_ch_tbl.c.changeid, sch_ch_tbl.c.important ],\n                whereclause=wc)\n            return dict([ (r.changeid, [False,True][r.important])\n                          for r in conn.execute(q) ])\n        return self.db.pool.do(thd)\n", "path": "master/buildbot/db/schedulers.py"}, {"content": "# This file is part of Buildbot.  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright Buildbot Team Members\n\nfrom twisted.internet import defer, reactor\nfrom twisted.python import log\nfrom buildbot import util, config\nfrom buildbot.util import NotABranch\nfrom collections import defaultdict\nfrom buildbot.changes import filter, changes\nfrom buildbot.schedulers import base, dependent\n\nclass BaseBasicScheduler(base.BaseScheduler):\n    \"\"\"\n    @param onlyImportant: If True, only important changes will be added to the\n                          buildset.\n    @type onlyImportant: boolean\n\n    \"\"\"\n\n    compare_attrs = (base.BaseScheduler.compare_attrs +\n                     ('treeStableTimer', 'change_filter', 'fileIsImportant',\n                      'onlyImportant') )\n\n    _reactor = reactor # for tests\n\n    fileIsImportant = None\n\n    class NotSet: pass\n    def __init__(self, name, shouldntBeSet=NotSet, treeStableTimer=None,\n                builderNames=None, branch=NotABranch, branches=NotABranch,\n                fileIsImportant=None, properties={}, categories=None,\n                change_filter=None, onlyImportant=False, **kwargs):\n        if shouldntBeSet is not self.NotSet:\n            config.error(\n                \"pass arguments to schedulers using keyword arguments\")\n        if fileIsImportant and not callable(fileIsImportant):\n            config.error(\n                \"fileIsImportant must be a callable\")\n\n        # initialize parent classes\n        base.BaseScheduler.__init__(self, name, builderNames, properties, **kwargs)\n\n        self.treeStableTimer = treeStableTimer\n        if fileIsImportant is not None:\n            self.fileIsImportant = fileIsImportant\n        self.onlyImportant = onlyImportant\n        self.change_filter = self.getChangeFilter(branch=branch,\n                branches=branches, change_filter=change_filter,\n                categories=categories)\n\n        # the IDelayedCall used to wake up when this scheduler's\n        # treeStableTimer expires.\n        self._stable_timers = defaultdict(lambda : None)\n        self._stable_timers_lock = defer.DeferredLock()\n\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        raise NotImplementedError\n\n    def startService(self, _returnDeferred=False):\n        base.BaseScheduler.startService(self)\n\n        d = self.startConsumingChanges(fileIsImportant=self.fileIsImportant,\n                                       change_filter=self.change_filter,\n                                       onlyImportant=self.onlyImportant)\n\n        # if treeStableTimer is False, then we don't care about classified\n        # changes, so get rid of any hanging around from previous\n        # configurations\n        if not self.treeStableTimer:\n            d.addCallback(lambda _ :\n                self.master.db.schedulers.flushChangeClassifications(\n                                                        self.objectid))\n\n        # otherwise, if there are classified changes out there, start their\n        # treeStableTimers again\n        else:\n            d.addCallback(lambda _ :\n                self.scanExistingClassifiedChanges())\n\n        # handle Deferred errors, since startService does not return a Deferred\n        d.addErrback(log.err, \"while starting SingleBranchScheduler '%s'\"\n                              % self.name)\n\n        if _returnDeferred:\n            return d # only used in tests\n\n    def stopService(self):\n        # the base stopService will unsubscribe from new changes\n        d = base.BaseScheduler.stopService(self)\n        @util.deferredLocked(self._stable_timers_lock)\n        def cancel_timers(_):\n            for timer in self._stable_timers.values():\n                if timer:\n                    timer.cancel()\n            self._stable_timers.clear()\n        d.addCallback(cancel_timers)\n        return d\n\n    @util.deferredLocked('_stable_timers_lock')\n    def gotChange(self, change, important):\n        if not self.treeStableTimer:\n            # if there's no treeStableTimer, we can completely ignore\n            # unimportant changes\n            if not important:\n                return defer.succeed(None)\n            # otherwise, we'll build it right away\n            return self.addBuildsetForChanges(reason='scheduler',\n                            changeids=[ change.number ])\n\n        timer_name = self.getTimerNameForChange(change)\n\n        # if we have a treeStableTimer, then record the change's importance\n        # and:\n        # - for an important change, start the timer\n        # - for an unimportant change, reset the timer if it is running\n        d = self.master.db.schedulers.classifyChanges(\n                self.objectid, { change.number : important })\n        def fix_timer(_):\n            if not important and not self._stable_timers[timer_name]:\n                return\n            if self._stable_timers[timer_name]:\n                self._stable_timers[timer_name].cancel()\n            def fire_timer():\n                d = self.stableTimerFired(timer_name)\n                d.addErrback(log.err, \"while firing stable timer\")\n            self._stable_timers[timer_name] = self._reactor.callLater(\n                    self.treeStableTimer, fire_timer)\n        d.addCallback(fix_timer)\n        return d\n\n    @defer.inlineCallbacks\n    def scanExistingClassifiedChanges(self):\n        # call gotChange for each classified change.  This is called at startup\n        # and is intended to re-start the treeStableTimer for any changes that\n        # had not yet been built when the scheduler was stopped.\n\n        # NOTE: this may double-call gotChange for changes that arrive just as\n        # the scheduler starts up.  In practice, this doesn't hurt anything.\n        classifications = \\\n                yield self.master.db.schedulers.getChangeClassifications(\n                                                                self.objectid)\n\n        # call gotChange for each change, after first fetching it from the db\n        for changeid, important in classifications.iteritems():\n            chdict = yield self.master.db.changes.getChange(changeid)\n\n            if not chdict:\n                continue\n\n            change = yield changes.Change.fromChdict(self.master, chdict)\n            yield self.gotChange(change, important)\n\n    def getTimerNameForChange(self, change):\n        raise NotImplementedError # see subclasses\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        \"\"\"similar to db.schedulers.getChangeClassifications, but given timer\n        name\"\"\"\n        raise NotImplementedError # see subclasses\n\n    @util.deferredLocked('_stable_timers_lock')\n    @defer.inlineCallbacks\n    def stableTimerFired(self, timer_name):\n        # if the service has already been stoppd then just bail out\n        if not self._stable_timers[timer_name]:\n            return\n\n        # delete this now-fired timer\n        del self._stable_timers[timer_name]\n\n        classifications = \\\n            yield self.getChangeClassificationsForTimer(self.objectid,\n                                                            timer_name)\n\n        # just in case: databases do weird things sometimes!\n        if not classifications: # pragma: no cover\n            return\n\n        changeids = sorted(classifications.keys())\n        yield self.addBuildsetForChanges(reason='scheduler',\n                                           changeids=changeids)\n\n        max_changeid = changeids[-1] # (changeids are sorted)\n        yield self.master.db.schedulers.flushChangeClassifications(\n                            self.objectid, less_than=max_changeid+1)\n\n    def getPendingBuildTimes(self):\n        # This isn't locked, since the caller expects and immediate value,\n        # and in any case, this is only an estimate.\n        return [timer.getTime() for timer in self._stable_timers.values() if timer and timer.active()]\n\nclass SingleBranchScheduler(BaseBasicScheduler):\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        if branch is NotABranch and not change_filter:\n            config.error(\n                \"the 'branch' argument to SingleBranchScheduler is \" +\n                \"mandatory unless change_filter is provided\")\n        elif branches is not NotABranch:\n            config.error(\n                \"the 'branches' argument is not allowed for \" +\n                \"SingleBranchScheduler\")\n\n\n        return filter.ChangeFilter.fromSchedulerConstructorArgs(\n                change_filter=change_filter, branch=branch,\n                categories=categories)\n\n    def getTimerNameForChange(self, change):\n        return \"only\" # this class only uses one timer\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        return self.master.db.schedulers.getChangeClassifications(\n                                                        self.objectid)\n\n\nclass Scheduler(SingleBranchScheduler):\n    \"alias for SingleBranchScheduler\"\n    def __init__(self, *args, **kwargs):\n        log.msg(\"WARNING: the name 'Scheduler' is deprecated; use \" +\n                \"buildbot.schedulers.basic.SingleBranchScheduler instead \" +\n                \"(note that this may require you to change your import \" +\n                \"statement)\")\n        SingleBranchScheduler.__init__(self, *args, **kwargs)\n\n\nclass AnyBranchScheduler(BaseBasicScheduler):\n    def getChangeFilter(self, branch, branches, change_filter, categories):\n        assert branch is NotABranch\n        return filter.ChangeFilter.fromSchedulerConstructorArgs(\n                change_filter=change_filter, branch=branches,\n                categories=categories)\n\n    def getTimerNameForChange(self, change):\n        # Py2.6+: could be a namedtuple\n        return (change.codebase, change.project, change.repository, change.branch)\n\n    def getChangeClassificationsForTimer(self, objectid, timer_name):\n        codebase, project, repository, branch = timer_name # set in getTimerNameForChange\n        return self.master.db.schedulers.getChangeClassifications(\n                self.objectid, branch=branch, repository=repository,\n                codebase=codebase, project=project)\n\n# now at buildbot.schedulers.dependent, but keep the old name alive\nDependent = dependent.Dependent\n", "path": "master/buildbot/schedulers/basic.py"}]}