Add logging for many pdu/edus

Merge pull request #2439 from matrix-org/rav/tox_tweaks
do tox install with pip -e
2017-09-11 16:28:20 +01:00 · 2017-09-06 17:08:19 +01:00 · 2017-09-05 21:58:19 +01:00 · 2017-09-05 21:57:19 +01:00 · 2017-09-05 17:51:13 +01:00 · 2017-09-05 16:35:23 +01:00
49 changed files with 1378 additions and 587 deletions
@@ -0,0 +1,47 @@
+<!-- 
+
+**IF YOU HAVE SUPPORT QUESTIONS ABOUT RUNNING OR CONFIGURING YOUR OWN HOME SERVER**: 
+You will likely get better support more quickly if you ask in ** #matrix:matrix.org ** ;)
+
+
+This is a bug report template. By following the instructions below and
+filling out the sections with your information, you will help the us to get all
+the necessary data to fix your issue.
+
+You can also preview your report before submitting it. You may remove sections
+that aren't relevant to your particular case.
+
+Text between <!-- and --> marks will be invisible in the report.
+
+-->
+
+### Description
+
+Describe here the problem that you are experiencing, or the feature you are requesting.
+
+### Steps to reproduce
+
+- For bugs, list the steps
+- that reproduce the bug
+- using hyphens as bullet points
+
+Describe how what happens differs from what you expected.
+
+If you can identify any relevant log snippets from _homeserver.log_, please include
+those here (please be careful to remove any personal or private data):
+
+### Version information
+
+<!-- IMPORTANT: please answer the following questions, to help us narrow down the problem -->
+
+- **Homeserver**: Was this issue identified on matrix.org or another homeserver?
+
+If not matrix.org:
+- **Version**:        What version of Synapse is running? <!-- 
+You can find the Synapse version by inspecting the server headers (replace matrix.org with
+your own homeserver domain):
+$ curl -v https://matrix.org/_matrix/client/versions 2>&1 | grep "Server:"
+-->
+- **Install method**: package manager/git clone/pip      
+- **Platform**:       Tell us about the environment in which your homeserver is operating
+                      - distro, hardware, if it's running in a vm/container, etc.
@@ -1,3 +1,18 @@
+Changes in synapse v0.22.1 (2017-07-06)
+=======================================
+
+Bug fixes:
+
+* Fix bug where pusher pool didn't start and caused issues when
+  interacting with some rooms (PR #2342)
+
+
+Changes in synapse v0.22.0 (2017-07-06)
+=======================================
+
+No changes since v0.22.0-rc2
+
+
 Changes in synapse v0.22.0-rc2 (2017-07-04)
 ===========================================

@@ -27,4 +27,5 @@ exclude jenkins*.sh
 exclude jenkins*
 recursive-exclude jenkins *.sh

+prune .github
 prune demo/etc
@@ -5,39 +5,48 @@ Before upgrading check if any special steps are required to upgrade from the
 what you currently have installed to current version of synapse. The extra
 instructions that may be required are listed later in this document.

-If synapse was installed in a virtualenv then active that virtualenv before
-upgrading. If synapse is installed in a virtualenv in ``~/.synapse/`` then run:
+1. If synapse was installed in a virtualenv then active that virtualenv before
+   upgrading. If synapse is installed in a virtualenv in ``~/.synapse/`` then
+   run:
+
+   .. code:: bash
+
+       source ~/.synapse/bin/activate
+
+2. If synapse was installed using pip then upgrade to the latest version by
+   running:
+
+   .. code:: bash
+
+       pip install --upgrade --process-dependency-links https://github.com/matrix-org/synapse/tarball/master
+
+       # restart synapse
+       synctl restart
+
+
+   If synapse was installed using git then upgrade to the latest version by
+   running:
+
+   .. code:: bash
+
+       # Pull the latest version of the master branch.
+       git pull
+       # Update the versions of synapse's python dependencies.
+       python synapse/python_dependencies.py | xargs pip install --upgrade
+
+       # restart synapse
+       ./synctl restart
+
+
+To check whether your update was sucessful, you can check the Server header
+returned by the Client-Server API:

 .. code:: bash

-    source ~/.synapse/bin/activate
-
-If synapse was installed using pip then upgrade to the latest version by
-running:
-
-.. code:: bash
-
-    pip install --upgrade --process-dependency-links https://github.com/matrix-org/synapse/tarball/master
-
-If synapse was installed using git then upgrade to the latest version by
-running:
-
-.. code:: bash
-
-    # Pull the latest version of the master branch.
-    git pull
-    # Update the versions of synapse's python dependencies.
-    python synapse/python_dependencies.py | xargs -n1 pip install --upgrade
-	
-To check whether your update was sucessfull, run:
-
-.. code:: bash
-
-	 # replace your.server.domain with ther domain of your synapse homeserver
-	 curl https://<your.server.domain>/_matrix/federation/v1/version 
-
-So for the Matrix.org HS server the URL would be: https://matrix.org/_matrix/federation/v1/version.
-
+    # replace <host.name> with the hostname of your synapse homeserver.
+    # You may need to specify a port (eg, :8448) if your server is not
+    # configured on port 443.
+    curl -kv https://<host.name>/_matrix/client/versions 2>&1 | grep "Server:"

 Upgrading to v0.15.0
 ====================
@@ -77,7 +86,7 @@ It has been replaced by specifying a list of application service registrations i
 ``homeserver.yaml``::

  app_service_config_files: ["registration-01.yaml", "registration-02.yaml"]
-  
+
 Where ``registration-01.yaml`` looks like::

  url: <String>  # e.g. "https://my.application.service.com"
@@ -166,7 +175,7 @@ This release completely changes the database schema and so requires upgrading
 it before starting the new version of the homeserver.

 The script "database-prepare-for-0.5.0.sh" should be used to upgrade the
-database. This will save all user information, such as logins and profiles, 
+database. This will save all user information, such as logins and profiles,
 but will otherwise purge the database. This includes messages, which
 rooms the home server was a member of and room alias mappings.

@@ -175,18 +184,18 @@ file and ask for help in #matrix:matrix.org. The upgrade process is,
 unfortunately, non trivial and requires human intervention to resolve any
 resulting conflicts during the upgrade process.

-Before running the command the homeserver should be first completely 
+Before running the command the homeserver should be first completely
 shutdown. To run it, simply specify the location of the database, e.g.:

  ./scripts/database-prepare-for-0.5.0.sh "homeserver.db"

-Once this has successfully completed it will be safe to restart the 
-homeserver. You may notice that the homeserver takes a few seconds longer to 
+Once this has successfully completed it will be safe to restart the
+homeserver. You may notice that the homeserver takes a few seconds longer to
 restart than usual as it reinitializes the database.

 On startup of the new version, users can either rejoin remote rooms using room
 aliases or by being reinvited. Alternatively, if any other homeserver sends a
-message to a room that the homeserver was previously in the local HS will 
+message to a room that the homeserver was previously in the local HS will
 automatically rejoin the room.

 Upgrading to v0.4.0
@@ -245,7 +254,7 @@ automatically generate default config use::
        --config-path homeserver.config \
        --generate-config

-This config can be edited if desired, for example to specify a different SSL 
+This config can be edited if desired, for example to specify a different SSL
 certificate to use. Once done you can run the home server using::

    $ python synapse/app/homeserver.py --config-path homeserver.config
@@ -266,20 +275,20 @@ This release completely changes the database schema and so requires upgrading
 it before starting the new version of the homeserver.

 The script "database-prepare-for-0.0.1.sh" should be used to upgrade the
-database. This will save all user information, such as logins and profiles, 
+database. This will save all user information, such as logins and profiles,
 but will otherwise purge the database. This includes messages, which
 rooms the home server was a member of and room alias mappings.

-Before running the command the homeserver should be first completely 
+Before running the command the homeserver should be first completely
 shutdown. To run it, simply specify the location of the database, e.g.:

  ./scripts/database-prepare-for-0.0.1.sh "homeserver.db"

-Once this has successfully completed it will be safe to restart the 
-homeserver. You may notice that the homeserver takes a few seconds longer to 
+Once this has successfully completed it will be safe to restart the
+homeserver. You may notice that the homeserver takes a few seconds longer to
 restart than usual as it reinitializes the database.

 On startup of the new version, users can either rejoin remote rooms using room
 aliases or by being reinvited. Alternatively, if any other homeserver sends a
-message to a room that the homeserver was previously in the local HS will 
+message to a room that the homeserver was previously in the local HS will
 automatically rejoin the room.
@@ -0,0 +1,20 @@
+This directory contains some sample monitoring config for using the
+'Prometheus' monitoring server against synapse.
+
+To use it, first install prometheus by following the instructions at
+
+  http://prometheus.io/
+
+Then add a new job to the main prometheus.conf file:
+
+  job: {
+    name: "synapse"
+
+    target_group: {
+      target: "http://SERVER.LOCATION.HERE:PORT/_synapse/metrics"
+    }
+  }
+
+Metrics are disabled by default when running synapse; they must be enabled
+with the 'enable-metrics' option, either in the synapse config file or as a
+command-line option.
@@ -0,0 +1,395 @@
+{{ template "head" . }}
+
+{{ template "prom_content_head" . }}
+<h1>System Resources</h1>
+
+<h3>CPU</h3>
+<div id="process_resource_utime"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#process_resource_utime"),
+  expr: "rate(process_cpu_seconds_total[2m]) * 100",
+  name: "[[job]]",  
+  min: 0,
+  max: 100,
+  renderer: "line",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "%",
+  yTitle: "CPU Usage"
+})
+</script>
+
+<h3>Memory</h3>
+<div id="process_resource_maxrss"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#process_resource_maxrss"),
+  expr: "process_psutil_rss:max",
+  name: "Maxrss",
+  min: 0,
+  renderer: "line",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "bytes",
+  yTitle: "Usage"
+})
+</script>
+
+<h3>File descriptors</h3>
+<div id="process_fds"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#process_fds"),
+  expr: "process_open_fds{job='synapse'}",
+  name: "FDs",
+  min: 0,
+  renderer: "line",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "",
+  yTitle: "Descriptors"
+})
+</script>
+
+<h1>Reactor</h1>
+
+<h3>Total reactor time</h3>
+<div id="reactor_total_time"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#reactor_total_time"),
+  expr: "rate(python_twisted_reactor_tick_time:total[2m]) / 1000",
+  name: "time",
+  max: 1,
+  min: 0,
+  renderer: "area",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/s",
+  yTitle: "Usage"
+})
+</script>
+
+<h3>Average reactor tick time</h3>
+<div id="reactor_average_time"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#reactor_average_time"),
+  expr: "rate(python_twisted_reactor_tick_time:total[2m]) / rate(python_twisted_reactor_tick_time:count[2m]) / 1000",
+  name: "time",
+  min: 0,
+  renderer: "line",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s",
+  yTitle: "Time"
+})
+</script>
+
+<h3>Pending calls per tick</h3>
+<div id="reactor_pending_calls"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#reactor_pending_calls"),
+  expr: "rate(python_twisted_reactor_pending_calls:total[30s])/rate(python_twisted_reactor_pending_calls:count[30s])",
+  name: "calls",
+  min: 0,
+  renderer: "line",
+  height: 150,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yTitle: "Pending Cals"
+})
+</script>
+
+<h1>Storage</h1>
+
+<h3>Queries</h3>
+<div id="synapse_storage_query_time"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_storage_query_time"),
+  expr: "rate(synapse_storage_query_time:count[2m])",
+  name: "[[verb]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "queries/s",
+  yTitle: "Queries"
+})
+</script>
+
+<h3>Transactions</h3>
+<div id="synapse_storage_transaction_time"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_storage_transaction_time"),
+  expr: "rate(synapse_storage_transaction_time:count[2m])",
+  name: "[[desc]]",
+  min: 0,
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "txn/s",
+  yTitle: "Transactions"
+})
+</script>
+
+<h3>Transaction execution time</h3>
+<div id="synapse_storage_transactions_time_msec"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_storage_transactions_time_msec"),
+  expr: "rate(synapse_storage_transaction_time:total[2m]) / 1000",
+  name: "[[desc]]",
+  min: 0,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/s",
+  yTitle: "Usage"
+})
+</script>
+
+<h3>Database scheduling latency</h3>
+<div id="synapse_storage_schedule_time"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_storage_schedule_time"),
+  expr: "rate(synapse_storage_schedule_time:total[2m]) / 1000",
+  name: "Total latency",
+  min: 0,
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/s",
+  yTitle: "Usage"
+})
+</script>
+
+<h3>Cache hit ratio</h3>
+<div id="synapse_cache_ratio"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_cache_ratio"),
+  expr: "rate(synapse_util_caches_cache:total[2m]) * 100",
+  name: "[[name]]",
+  min: 0,
+  max: 100,
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "%",
+  yTitle: "Percentage"
+})
+</script>
+
+<h3>Cache size</h3>
+<div id="synapse_cache_size"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_cache_size"),
+  expr: "synapse_util_caches_cache:size",
+  name: "[[name]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "",
+  yTitle: "Items"
+})
+</script>
+
+<h1>Requests</h1>
+
+<h3>Requests by Servlet</h3>
+<div id="synapse_http_server_requests_servlet"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_requests_servlet"),
+  expr: "rate(synapse_http_server_requests:servlet[2m])",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+<h4>&nbsp;(without <tt>EventStreamRestServlet</tt> or <tt>SyncRestServlet</tt>)</h4>
+<div id="synapse_http_server_requests_servlet_minus_events"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_requests_servlet_minus_events"),
+  expr: "rate(synapse_http_server_requests:servlet{servlet!=\"EventStreamRestServlet\", servlet!=\"SyncRestServlet\"}[2m])",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+
+<h3>Average response times</h3>
+<div id="synapse_http_server_response_time_avg"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_response_time_avg"),
+  expr: "rate(synapse_http_server_response_time:total[2m]) / rate(synapse_http_server_response_time:count[2m]) / 1000",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/req",
+  yTitle: "Response time"
+})
+</script>
+
+<h3>All responses by code</h3>
+<div id="synapse_http_server_responses"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_responses"),
+  expr: "rate(synapse_http_server_responses[2m])",
+  name: "[[method]] / [[code]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+
+<h3>Error responses by code</h3>
+<div id="synapse_http_server_responses_err"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_responses_err"),
+  expr: "rate(synapse_http_server_responses{code=~\"[45]..\"}[2m])",
+  name: "[[method]] / [[code]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+
+
+<h3>CPU Usage</h3>
+<div id="synapse_http_server_response_ru_utime"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_response_ru_utime"),
+  expr: "rate(synapse_http_server_response_ru_utime:total[2m])",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/s",
+  yTitle: "CPU Usage"
+})
+</script>
+
+
+<h3>DB Usage</h3>
+<div id="synapse_http_server_response_db_txn_duration"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_response_db_txn_duration"),
+  expr: "rate(synapse_http_server_response_db_txn_duration:total[2m])",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/s",
+  yTitle: "DB Usage"
+})
+</script>
+
+
+<h3>Average event send times</h3>
+<div id="synapse_http_server_send_time_avg"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_http_server_send_time_avg"),
+  expr: "rate(synapse_http_server_response_time:total{servlet='RoomSendEventRestServlet'}[2m]) / rate(synapse_http_server_response_time:count{servlet='RoomSendEventRestServlet'}[2m]) / 1000",
+  name: "[[servlet]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "s/req",
+  yTitle: "Response time"
+})
+</script>
+
+<h1>Federation</h1>
+
+<h3>Sent Messages</h3>
+<div id="synapse_federation_client_sent"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_federation_client_sent"),
+  expr: "rate(synapse_federation_client_sent[2m])",
+  name: "[[type]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+
+<h3>Received Messages</h3>
+<div id="synapse_federation_server_received"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_federation_server_received"),
+  expr: "rate(synapse_federation_server_received[2m])",
+  name: "[[type]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "req/s",
+  yTitle: "Requests"
+})
+</script>
+
+<h3>Pending</h3>
+<div id="synapse_federation_transaction_queue_pending"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_federation_transaction_queue_pending"),
+  expr: "synapse_federation_transaction_queue_pending",
+  name: "[[type]]",
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "",
+  yTitle: "Units"
+})
+</script>
+
+<h1>Clients</h1>
+
+<h3>Notifiers</h3>
+<div id="synapse_notifier_listeners"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_notifier_listeners"),
+  expr: "synapse_notifier_listeners",
+  name: "listeners",
+  min: 0,
+  yAxisFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yHoverFormatter: PromConsole.NumberFormatter.humanizeNoSmallPrefix,
+  yUnits: "",
+  yTitle: "Listeners"
+})
+</script>
+
+<h3>Notified Events</h3>
+<div id="synapse_notifier_notified_events"></div>
+<script>
+new PromConsole.Graph({
+  node: document.querySelector("#synapse_notifier_notified_events"),
+  expr: "rate(synapse_notifier_notified_events[2m])",
+  name: "events",
+  yAxisFormatter: PromConsole.NumberFormatter.humanize,
+  yHoverFormatter: PromConsole.NumberFormatter.humanize,
+  yUnits: "events/s",
+  yTitle: "Event rate"
+})
+</script>
+
+{{ template "prom_content_tail" . }}
+
+{{ template "tail" }}
@@ -0,0 +1,21 @@
+synapse_federation_transaction_queue_pendingEdus:total = sum(synapse_federation_transaction_queue_pendingEdus or absent(synapse_federation_transaction_queue_pendingEdus)*0)
+synapse_federation_transaction_queue_pendingPdus:total = sum(synapse_federation_transaction_queue_pendingPdus or absent(synapse_federation_transaction_queue_pendingPdus)*0)
+
+synapse_http_server_requests:method{servlet=""} = sum(synapse_http_server_requests) by (method)
+synapse_http_server_requests:servlet{method=""} = sum(synapse_http_server_requests) by (servlet)
+
+synapse_http_server_requests:total{servlet=""} = sum(synapse_http_server_requests:by_method) by (servlet)
+
+synapse_cache:hit_ratio_5m = rate(synapse_util_caches_cache:hits[5m]) / rate(synapse_util_caches_cache:total[5m])
+synapse_cache:hit_ratio_30s = rate(synapse_util_caches_cache:hits[30s]) / rate(synapse_util_caches_cache:total[30s])
+
+synapse_federation_client_sent{type="EDU"} = synapse_federation_client_sent_edus + 0
+synapse_federation_client_sent{type="PDU"} = synapse_federation_client_sent_pdu_destinations:count + 0
+synapse_federation_client_sent{type="Query"} = sum(synapse_federation_client_sent_queries) by (job)
+
+synapse_federation_server_received{type="EDU"} = synapse_federation_server_received_edus + 0
+synapse_federation_server_received{type="PDU"} = synapse_federation_server_received_pdus + 0
+synapse_federation_server_received{type="Query"} = sum(synapse_federation_server_received_queries) by (job)
+
+synapse_federation_transaction_queue_pending{type="EDU"} = synapse_federation_transaction_queue_pending_edus + 0
+synapse_federation_transaction_queue_pending{type="PDU"} = synapse_federation_transaction_queue_pending_pdus + 0
@@ -1,6 +1,8 @@
 Using Postgres
 --------------

+Postgres version 9.4 or later is known to work.
+
 Set up database
 ===============

@@ -17,6 +17,7 @@ export HAPROXY_BIN=/home/haproxy/haproxy-1.6.11/haproxy
 ./sytest/jenkins/prep_sytest_for_postgres.sh

 ./sytest/jenkins/install_and_run.sh \
+    --python $WORKSPACE/.tox/py27/bin/python \
    --synapse-directory $WORKSPACE \
    --dendron $WORKSPACE/dendron/bin/dendron \
    --haproxy \
@@ -15,5 +15,6 @@ export SYNAPSE_CACHE_FACTOR=1
 ./sytest/jenkins/prep_sytest_for_postgres.sh

 ./sytest/jenkins/install_and_run.sh \
+    --python $WORKSPACE/.tox/py27/bin/python \
    --synapse-directory $WORKSPACE \
    --dendron $WORKSPACE/dendron/bin/dendron \
@@ -14,4 +14,5 @@ export SYNAPSE_CACHE_FACTOR=1
 ./sytest/jenkins/prep_sytest_for_postgres.sh

 ./sytest/jenkins/install_and_run.sh \
+    --python $WORKSPACE/.tox/py27/bin/python \
    --synapse-directory $WORKSPACE \
@@ -12,4 +12,5 @@ export SYNAPSE_CACHE_FACTOR=1
 ./jenkins/clone.sh sytest https://github.com/matrix-org/sytest.git

 ./sytest/jenkins/install_and_run.sh \
+    --python $WORKSPACE/.tox/py27/bin/python \
    --synapse-directory $WORKSPACE \
@@ -1,10 +1,30 @@
+#!/usr/bin/env python
+#
+# Copyright 2015, 2016 OpenMarket Ltd
+# Copyright 2017 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import print_function
+
+import argparse
 import nacl.signing
 import json
 import base64
 import requests
 import sys
 import srvlookup
-
+import yaml

 def encode_base64(input_bytes):
    """Encode bytes as a base64 string without any padding."""
@@ -120,11 +140,13 @@ def get_json(origin_name, origin_key, destination, path):
            origin_name, key, sig,
        )
        authorization_headers.append(bytes(header))
-        sys.stderr.write(header)
-        sys.stderr.write("\n")
+        print ("Authorization: %s" % header, file=sys.stderr)
+
+    dest = lookup(destination, path)
+    print ("Requesting %s" % dest, file=sys.stderr)

    result = requests.get(
-        lookup(destination, path),
+        dest,
        headers={"Authorization": authorization_headers[0]},
        verify=False,
    )
@@ -133,17 +155,66 @@ def get_json(origin_name, origin_key, destination, path):


 def main():
-    origin_name, keyfile, destination, path = sys.argv[1:]
+    parser = argparse.ArgumentParser(
+        description=
+            "Signs and sends a federation request to a matrix homeserver",
+    )

-    with open(keyfile) as f:
+    parser.add_argument(
+        "-N", "--server-name",
+        help="Name to give as the local homeserver. If unspecified, will be "
+             "read from the config file.",
+    )
+
+    parser.add_argument(
+        "-k", "--signing-key-path",
+        help="Path to the file containing the private ed25519 key to sign the "
+             "request with.",
+    )
+
+    parser.add_argument(
+        "-c", "--config",
+        default="homeserver.yaml",
+        help="Path to server config file. Ignored if --server-name and "
+             "--signing-key-path are both given.",
+    )
+
+    parser.add_argument(
+        "-d", "--destination",
+        default="matrix.org",
+        help="name of the remote homeserver. We will do SRV lookups and "
+             "connect appropriately.",
+    )
+
+    parser.add_argument(
+        "path",
+        help="request path. We will add '/_matrix/federation/v1/' to this."
+    )
+
+    args = parser.parse_args()
+
+    if not args.server_name or not args.signing_key_path:
+        read_args_from_config(args)
+
+    with open(args.signing_key_path) as f:
        key = read_signing_keys(f)[0]

    result = get_json(
-        origin_name, key, destination, "/_matrix/federation/v1/" + path
+        args.server_name, key, args.destination, "/_matrix/federation/v1/" + args.path
    )

    json.dump(result, sys.stdout)
-    print ""
+    print ("")
+
+
+def read_args_from_config(args):
+    with open(args.config, 'r') as fh:
+        config = yaml.safe_load(fh)
+        if not args.server_name:
+            args.server_name = config['server_name']
+        if not args.signing_key_path:
+            args.signing_key_path = config['signing_key_path']
+

 if __name__ == "__main__":
    main()
@@ -252,6 +252,25 @@ class Porter(object):
            )
            return

+        if table in (
+            "user_directory", "user_directory_search", "users_who_share_rooms",
+            "users_in_pubic_room",
+        ):
+            # We don't port these tables, as they're a faff and we can regenreate
+            # them anyway.
+            self.progress.update(table, table_size)  # Mark table as done
+            return
+
+        if table == "user_directory_stream_pos":
+            # We need to make sure there is a single row, `(X, null), as that is
+            # what synapse expects to be there.
+            yield self.postgres_store._simple_insert(
+                table=table,
+                values={"stream_id": None},
+            )
+            self.progress.update(table, table_size)  # Mark table as done
+            return
+
        forward_select = (
            "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?"
            % (table,)
@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """

-__version__ = "0.22.0-rc2"
+__version__ = "0.22.1"
@@ -209,7 +209,7 @@ class Auth(object):
            )[0]
            if user and access_token and ip_addr:
                self.store.insert_client_ip(
-                    user=user,
+                    user_id=user.to_string(),
                    access_token=access_token,
                    ip=ip_addr,
                    user_agent=user_agent,
@@ -0,0 +1,99 @@
+# -*- coding: utf-8 -*-
+# Copyright 2017 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import gc
+import logging
+
+import affinity
+from daemonize import Daemonize
+from synapse.util import PreserveLoggingContext
+from synapse.util.rlimit import change_resource_limit
+from twisted.internet import reactor
+
+
+def start_worker_reactor(appname, config):
+    """ Run the reactor in the main process
+
+    Daemonizes if necessary, and then configures some resources, before starting
+    the reactor. Pulls configuration from the 'worker' settings in 'config'.
+
+    Args:
+        appname (str): application name which will be sent to syslog
+        config (synapse.config.Config): config object
+    """
+
+    logger = logging.getLogger(config.worker_app)
+
+    start_reactor(
+        appname,
+        config.soft_file_limit,
+        config.gc_thresholds,
+        config.worker_pid_file,
+        config.worker_daemonize,
+        config.worker_cpu_affinity,
+        logger,
+    )
+
+
+def start_reactor(
+        appname,
+        soft_file_limit,
+        gc_thresholds,
+        pid_file,
+        daemonize,
+        cpu_affinity,
+        logger,
+):
+    """ Run the reactor in the main process
+
+    Daemonizes if necessary, and then configures some resources, before starting
+    the reactor
+
+    Args:
+        appname (str): application name which will be sent to syslog
+        soft_file_limit (int):
+        gc_thresholds:
+        pid_file (str): name of pid file to write to if daemonize is True
+        daemonize (bool): true to run the reactor in a background process
+        cpu_affinity (int|None): cpu affinity mask
+        logger (logging.Logger): logger instance to pass to Daemonize
+    """
+
+    def run():
+        # make sure that we run the reactor with the sentinel log context,
+        # otherwise other PreserveLoggingContext instances will get confused
+        # and complain when they see the logcontext arbitrarily swapping
+        # between the sentinel and `run` logcontexts.
+        with PreserveLoggingContext():
+            logger.info("Running")
+            if cpu_affinity is not None:
+                logger.info("Setting CPU affinity to %s" % cpu_affinity)
+                affinity.set_process_affinity_mask(0, cpu_affinity)
+            change_resource_limit(soft_file_limit)
+            if gc_thresholds:
+                gc.set_threshold(*gc_thresholds)
+            reactor.run()
+
+    if daemonize:
+        daemon = Daemonize(
+            app=appname,
+            pid=pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
@@ -13,38 +13,31 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
-from synapse.server import HomeServer
+from synapse import events
+from synapse.app import _base
 from synapse.config._base import ConfigError
-from synapse.config.logger import setup_logging
 from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
 from synapse.http.site import SynapseSite
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.events import SlavedEventStore
-from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.tcp.client import ReplicationClientHandler
+from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext, preserve_fn
+from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-
-from synapse import events
-
 from twisted.internet import reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.appservice")


@@ -181,36 +174,13 @@ def start(config_options):
    ps.setup()
    ps.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ps.get_datastore().start_profiling()
        ps.get_state_handler().start_caching()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-appservice",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-appservice", config)


 if __name__ == '__main__':
@@ -13,47 +13,39 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
+from synapse import events
+from synapse.app import _base
 from synapse.config._base import ConfigError
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.logger import setup_logging
-from synapse.http.site import SynapseSite
+from synapse.crypto import context_factory
 from synapse.http.server import JsonResource
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
+from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.keys import SlavedKeyStore
-from synapse.replication.slave.storage.room import RoomStore
-from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.replication.slave.storage.room import RoomStore
 from synapse.replication.slave.storage.transactions import TransactionStore
 from synapse.replication.tcp.client import ReplicationClientHandler
 from synapse.rest.client.v1.room import PublicRoomListRestServlet
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
+from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-from synapse.crypto import context_factory
-
-from synapse import events
-
-
 from twisted.internet import reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.client_reader")


@@ -183,36 +175,13 @@ def start(config_options):
    ss.get_handlers()
    ss.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ss.get_state_handler().start_caching()
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-client-reader",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-client-reader", config)


 if __name__ == '__main__':
@@ -13,44 +13,36 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
+from synapse import events
+from synapse.api.urls import FEDERATION_PREFIX
+from synapse.app import _base
 from synapse.config._base import ConfigError
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.logger import setup_logging
+from synapse.crypto import context_factory
+from synapse.federation.transport.server import TransportLayerServer
 from synapse.http.site import SynapseSite
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.keys import SlavedKeyStore
 from synapse.replication.slave.storage.room import RoomStore
 from synapse.replication.slave.storage.transactions import TransactionStore
-from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.tcp.client import ReplicationClientHandler
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
+from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-from synapse.api.urls import FEDERATION_PREFIX
-from synapse.federation.transport.server import TransportLayerServer
-from synapse.crypto import context_factory
-
-from synapse import events
-
-
 from twisted.internet import reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.federation_reader")


@@ -172,36 +164,13 @@ def start(config_options):
    ss.get_handlers()
    ss.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ss.get_state_handler().start_caching()
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-federation-reader",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-federation-reader", config)


 if __name__ == '__main__':
@@ -13,44 +13,37 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
-from synapse.server import HomeServer
+from synapse import events
+from synapse.app import _base
 from synapse.config._base import ConfigError
-from synapse.config.logger import setup_logging
 from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
 from synapse.crypto import context_factory
-from synapse.http.site import SynapseSite
 from synapse.federation import send_queue
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage.deviceinbox import SlavedDeviceInboxStore
+from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.presence import SlavedPresenceStore
 from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
-from synapse.replication.slave.storage.presence import SlavedPresenceStore
 from synapse.replication.slave.storage.transactions import TransactionStore
-from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.tcp.client import ReplicationClientHandler
+from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.async import Linearizer
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext, preserve_fn
+from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-
-from synapse import events
-
-from twisted.internet import reactor, defer
+from twisted.internet import defer, reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.federation_sender")


@@ -213,36 +206,12 @@ def start(config_options):
    ps.setup()
    ps.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ps.get_datastore().start_profiling()
        ps.get_state_handler().start_caching()

    reactor.callWhenRunning(start)
-
-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-federation-sender",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-federation-sender", config)


 class FederationSenderHandler(object):
@@ -0,0 +1,239 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import sys
+
+import synapse
+from synapse import events
+from synapse.api.errors import SynapseError
+from synapse.app import _base
+from synapse.config._base import ConfigError
+from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
+from synapse.crypto import context_factory
+from synapse.http.server import JsonResource
+from synapse.http.servlet import (
+    RestServlet, parse_json_object_from_request,
+)
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
+from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
+from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
+from synapse.replication.slave.storage.devices import SlavedDeviceStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.replication.tcp.client import ReplicationClientHandler
+from synapse.rest.client.v2_alpha._base import client_v2_patterns
+from synapse.server import HomeServer
+from synapse.storage.engines import create_engine
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
+from synapse.util.versionstring import get_version_string
+from twisted.internet import defer, reactor
+from twisted.web.resource import Resource
+
+logger = logging.getLogger("synapse.app.frontend_proxy")
+
+
+class KeyUploadServlet(RestServlet):
+    PATTERNS = client_v2_patterns("/keys/upload(/(?P<device_id>[^/]+))?$",
+                                  releases=())
+
+    def __init__(self, hs):
+        """
+        Args:
+            hs (synapse.server.HomeServer): server
+        """
+        super(KeyUploadServlet, self).__init__()
+        self.auth = hs.get_auth()
+        self.store = hs.get_datastore()
+        self.http_client = hs.get_simple_http_client()
+        self.main_uri = hs.config.worker_main_http_uri
+
+    @defer.inlineCallbacks
+    def on_POST(self, request, device_id):
+        requester = yield self.auth.get_user_by_req(request, allow_guest=True)
+        user_id = requester.user.to_string()
+        body = parse_json_object_from_request(request)
+
+        if device_id is not None:
+            # passing the device_id here is deprecated; however, we allow it
+            # for now for compatibility with older clients.
+            if (requester.device_id is not None and
+                    device_id != requester.device_id):
+                logger.warning("Client uploading keys for a different device "
+                               "(logged in as %s, uploading for %s)",
+                               requester.device_id, device_id)
+        else:
+            device_id = requester.device_id
+
+        if device_id is None:
+            raise SynapseError(
+                400,
+                "To upload keys, you must pass device_id when authenticating"
+            )
+
+        if body:
+            # They're actually trying to upload something, proxy to main synapse.
+            result = yield self.http_client.post_json_get_json(
+                self.main_uri + request.uri,
+                body,
+            )
+
+            defer.returnValue((200, result))
+        else:
+            # Just interested in counts.
+            result = yield self.store.count_e2e_one_time_keys(user_id, device_id)
+            defer.returnValue((200, {"one_time_key_counts": result}))
+
+
+class FrontendProxySlavedStore(
+    SlavedDeviceStore,
+    SlavedClientIpStore,
+    SlavedApplicationServiceStore,
+    SlavedRegistrationStore,
+    BaseSlavedStore,
+):
+    pass
+
+
+class FrontendProxyServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = FrontendProxySlavedStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_addresses = listener_config["bind_addresses"]
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+                elif name == "client":
+                    resource = JsonResource(self, canonical_json=False)
+                    KeyUploadServlet(self).register(resource)
+                    resources.update({
+                        "/_matrix/client/r0": resource,
+                        "/_matrix/client/unstable": resource,
+                        "/_matrix/client/v2_alpha": resource,
+                        "/_matrix/client/api/v1": resource,
+                    })
+
+        root_resource = create_resource_tree(resources, Resource())
+
+        for address in bind_addresses:
+            reactor.listenTCP(
+                port,
+                SynapseSite(
+                    "synapse.access.http.%s" % (site_tag,),
+                    site_tag,
+                    listener_config,
+                    root_resource,
+                ),
+                interface=address
+            )
+
+        logger.info("Synapse client reader now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                bind_addresses = listener["bind_addresses"]
+
+                for address in bind_addresses:
+                    reactor.listenTCP(
+                        listener["port"],
+                        manhole(
+                            username="matrix",
+                            password="rabbithole",
+                            globals={"hs": self},
+                        ),
+                        interface=address
+                    )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+        self.get_tcp_replication().start_replication(self)
+
+    def build_tcp_replication(self):
+        return ReplicationClientHandler(self.get_datastore())
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse frontend proxy", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.frontend_proxy"
+
+    assert config.worker_main_http_uri is not None
+
+    setup_logging(config, use_worker_options=True)
+
+    events.USE_FROZEN_DICTS = config.use_frozen_dicts
+
+    database_engine = create_engine(config.database_config)
+
+    tls_server_context_factory = context_factory.ServerContextFactory(config)
+
+    ss = FrontendProxyServer(
+        config.server_name,
+        db_config=config.database_config,
+        tls_server_context_factory=tls_server_context_factory,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ss.setup()
+    ss.get_handlers()
+    ss.start_listening(config.worker_listeners)
+
+    def start():
+        ss.get_state_handler().start_caching()
+        ss.get_datastore().start_profiling()
+
+    reactor.callWhenRunning(start)
+
+    _base.start_worker_reactor("synapse-frontend-proxy", config)
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])
@@ -13,61 +13,48 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-import synapse
-
 import gc
 import logging
 import os
 import sys

+import synapse
 import synapse.config.logger
+from synapse import events
+from synapse.api.urls import CONTENT_REPO_PREFIX, FEDERATION_PREFIX, \
+    LEGACY_MEDIA_PREFIX, MEDIA_PREFIX, SERVER_KEY_PREFIX, SERVER_KEY_V2_PREFIX, \
+    STATIC_PREFIX, WEB_CLIENT_PREFIX
+from synapse.app import _base
 from synapse.config._base import ConfigError
-
-from synapse.python_dependencies import (
-    check_requirements, CONDITIONAL_REQUIREMENTS
-)
-
-from synapse.rest import ClientRestResource
-from synapse.storage.engines import create_engine, IncorrectDatabaseSetup
-from synapse.storage import are_all_users_on_domain
-from synapse.storage.prepare_database import UpgradeDatabaseException, prepare_database
-
-from synapse.server import HomeServer
-
-from twisted.internet import reactor, defer
-from twisted.application import service
-from twisted.web.resource import Resource, EncodingResourceWrapper
-from twisted.web.static import File
-from twisted.web.server import GzipEncoderFactory
-from synapse.http.server import RootRedirect
-from synapse.rest.media.v0.content_repository import ContentRepoResource
-from synapse.rest.media.v1.media_repository import MediaRepositoryResource
-from synapse.rest.key.v1.server_key_resource import LocalKey
-from synapse.rest.key.v2 import KeyApiV2Resource
-from synapse.api.urls import (
-    FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX,
-    SERVER_KEY_PREFIX, LEGACY_MEDIA_PREFIX, MEDIA_PREFIX, STATIC_PREFIX,
-    SERVER_KEY_V2_PREFIX,
-)
 from synapse.config.homeserver import HomeServerConfig
 from synapse.crypto import context_factory
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
-from synapse.metrics import register_memory_metrics
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
-from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory
 from synapse.federation.transport.server import TransportLayerServer
-
+from synapse.http.server import RootRedirect
+from synapse.http.site import SynapseSite
+from synapse.metrics import register_memory_metrics
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
+from synapse.python_dependencies import CONDITIONAL_REQUIREMENTS, \
+    check_requirements
+from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory
+from synapse.rest import ClientRestResource
+from synapse.rest.key.v1.server_key_resource import LocalKey
+from synapse.rest.key.v2 import KeyApiV2Resource
+from synapse.rest.media.v0.content_repository import ContentRepoResource
+from synapse.rest.media.v1.media_repository import MediaRepositoryResource
+from synapse.server import HomeServer
+from synapse.storage import are_all_users_on_domain
+from synapse.storage.engines import IncorrectDatabaseSetup, create_engine
+from synapse.storage.prepare_database import UpgradeDatabaseException, prepare_database
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
 from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.manhole import manhole
-
-from synapse.http.site import SynapseSite
-
-from synapse import events
-
-from daemonize import Daemonize
+from twisted.application import service
+from twisted.internet import defer, reactor
+from twisted.web.resource import EncodingResourceWrapper, Resource
+from twisted.web.server import GzipEncoderFactory
+from twisted.web.static import File

 logger = logging.getLogger("synapse.app.homeserver")

@@ -446,37 +433,18 @@ def run(hs):
        # be quite busy the first few minutes
        clock.call_later(5 * 60, phone_stats_home)

-    def in_thread():
-        # Uncomment to enable tracing of log context changes.
-        # sys.settrace(logcontext_tracer)
+    if hs.config.daemonize and hs.config.print_pidfile:
+        print (hs.config.pid_file)

-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            change_resource_limit(hs.config.soft_file_limit)
-            if hs.config.gc_thresholds:
-                gc.set_threshold(*hs.config.gc_thresholds)
-            reactor.run()
-
-    if hs.config.daemonize:
-
-        if hs.config.print_pidfile:
-            print (hs.config.pid_file)
-
-        daemon = Daemonize(
-            app="synapse-homeserver",
-            pid=hs.config.pid_file,
-            action=lambda: in_thread(),
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-
-        daemon.start()
-    else:
-        in_thread()
+    _base.start_reactor(
+        "synapse-homeserver",
+        hs.config.soft_file_limit,
+        hs.config.gc_thresholds,
+        hs.config.pid_file,
+        hs.config.daemonize,
+        hs.config.cpu_affinity,
+        logger,
+    )


 def main():
@@ -13,14 +13,21 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
+from synapse import events
+from synapse.api.urls import (
+    CONTENT_REPO_PREFIX, LEGACY_MEDIA_PREFIX, MEDIA_PREFIX
+)
+from synapse.app import _base
 from synapse.config._base import ConfigError
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.logger import setup_logging
+from synapse.crypto import context_factory
 from synapse.http.site import SynapseSite
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
@@ -33,27 +40,12 @@ from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.storage.media_repository import MediaRepositoryStore
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
+from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-from synapse.api.urls import (
-    CONTENT_REPO_PREFIX, LEGACY_MEDIA_PREFIX, MEDIA_PREFIX
-)
-from synapse.crypto import context_factory
-
-from synapse import events
-
-
 from twisted.internet import reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.media_repository")


@@ -180,36 +172,13 @@ def start(config_options):
    ss.get_handlers()
    ss.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ss.get_state_handler().start_caching()
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-media-repository",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-media-repository", config)


 if __name__ == '__main__':
@@ -13,41 +13,33 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
+import sys

 import synapse
-
-from synapse.server import HomeServer
+from synapse import events
+from synapse.app import _base
 from synapse.config._base import ConfigError
-from synapse.config.logger import setup_logging
 from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
 from synapse.http.site import SynapseSite
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
-from synapse.storage.roommember import RoomMemberStore
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
+from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.pushers import SlavedPusherStore
 from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
-from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.replication.tcp.client import ReplicationClientHandler
-from synapse.storage.engines import create_engine
+from synapse.server import HomeServer
 from synapse.storage import DataStore
+from synapse.storage.engines import create_engine
+from synapse.storage.roommember import RoomMemberStore
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, preserve_fn, \
-    PreserveLoggingContext
+from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
-
-from synapse import events
-
-from twisted.internet import reactor, defer
+from twisted.internet import defer, reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.pusher")


@@ -244,18 +236,6 @@ def start(config_options):
    ps.setup()
    ps.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ps.get_pusherpool().start()
        ps.get_datastore().start_profiling()
@@ -263,18 +243,7 @@ def start(config_options):

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-pusher",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-pusher", config)


 if __name__ == '__main__':
@@ -13,56 +13,50 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import contextlib
+import logging
+import sys

 import synapse
-
 from synapse.api.constants import EventTypes
+from synapse.app import _base
 from synapse.config._base import ConfigError
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.logger import setup_logging
 from synapse.handlers.presence import PresenceHandler, get_interested_parties
-from synapse.http.site import SynapseSite
 from synapse.http.server import JsonResource
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
-from synapse.rest.client.v2_alpha import sync
-from synapse.rest.client.v1 import events
-from synapse.rest.client.v1.room import RoomInitialSyncRestServlet
-from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
-from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
-from synapse.replication.slave.storage.events import SlavedEventStore
-from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
-from synapse.replication.slave.storage.registration import SlavedRegistrationStore
-from synapse.replication.slave.storage.filtering import SlavedFilteringStore
-from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
-from synapse.replication.slave.storage.presence import SlavedPresenceStore
+from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
 from synapse.replication.slave.storage.deviceinbox import SlavedDeviceInboxStore
 from synapse.replication.slave.storage.devices import SlavedDeviceStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.filtering import SlavedFilteringStore
+from synapse.replication.slave.storage.presence import SlavedPresenceStore
+from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
+from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.slave.storage.room import RoomStore
 from synapse.replication.tcp.client import ReplicationClientHandler
+from synapse.rest.client.v1 import events
+from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
+from synapse.rest.client.v1.room import RoomInitialSyncRestServlet
+from synapse.rest.client.v2_alpha import sync
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.storage.presence import UserPresenceState
 from synapse.storage.roommember import RoomMemberStore
 from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext, preserve_fn
+from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
 from synapse.util.stringutils import random_string
 from synapse.util.versionstring import get_version_string
-
-from twisted.internet import reactor, defer
+from twisted.internet import defer, reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import contextlib
-import gc
-
 logger = logging.getLogger("synapse.app.synchrotron")


@@ -440,36 +434,13 @@ def start(config_options):
    ss.setup()
    ss.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ss.get_datastore().start_profiling()
        ss.get_state_handler().start_caching()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-synchrotron",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-synchrotron", config)


 if __name__ == '__main__':
@@ -14,16 +14,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-import synapse
+import logging
+import sys

-from synapse.server import HomeServer
+import synapse
+from synapse import events
+from synapse.app import _base
 from synapse.config._base import ConfigError
-from synapse.config.logger import setup_logging
 from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
 from synapse.crypto import context_factory
-from synapse.http.site import SynapseSite
 from synapse.http.server import JsonResource
-from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
@@ -31,26 +34,17 @@ from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.tcp.client import ReplicationClientHandler
 from synapse.rest.client.v2_alpha import user_directory
+from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.storage.user_directory import UserDirectoryStore
-from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext, PreserveLoggingContext, preserve_fn
-from synapse.util.manhole import manhole
-from synapse.util.rlimit import change_resource_limit
-from synapse.util.versionstring import get_version_string
 from synapse.util.caches.stream_change_cache import StreamChangeCache
-
-from synapse import events
-
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext, preserve_fn
+from synapse.util.manhole import manhole
+from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
 from twisted.web.resource import Resource

-from daemonize import Daemonize
-
-import sys
-import logging
-import gc
-
 logger = logging.getLogger("synapse.app.user_dir")


@@ -233,36 +227,13 @@ def start(config_options):
    ps.setup()
    ps.start_listening(config.worker_listeners)

-    def run():
-        # make sure that we run the reactor with the sentinel log context,
-        # otherwise other PreserveLoggingContext instances will get confused
-        # and complain when they see the logcontext arbitrarily swapping
-        # between the sentinel and `run` logcontexts.
-        with PreserveLoggingContext():
-            logger.info("Running")
-            change_resource_limit(config.soft_file_limit)
-            if config.gc_thresholds:
-                gc.set_threshold(*config.gc_thresholds)
-            reactor.run()
-
    def start():
        ps.get_datastore().start_profiling()
        ps.get_state_handler().start_caching()

    reactor.callWhenRunning(start)

-    if config.worker_daemonize:
-        daemon = Daemonize(
-            app="synapse-user-dir",
-            pid=config.worker_pid_file,
-            action=run,
-            auto_close_fds=False,
-            verbose=True,
-            logger=logger,
-        )
-        daemon.start()
-    else:
-        run()
+    _base.start_worker_reactor("synapse-user-dir", config)


 if __name__ == '__main__':
@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
+# Copyright 2017 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -29,6 +30,7 @@ class ServerConfig(Config):
        self.user_agent_suffix = config.get("user_agent_suffix")
        self.use_frozen_dicts = config.get("use_frozen_dicts", False)
        self.public_baseurl = config.get("public_baseurl")
+        self.cpu_affinity = config.get("cpu_affinity")

        # Whether to send federation traffic out in this process. This only
        # applies to some federation traffic, and so shouldn't be used to
@@ -147,6 +149,27 @@ class ServerConfig(Config):
        # When running as a daemon, the file to store the pid in
        pid_file: %(pid_file)s

+        # CPU affinity mask. Setting this restricts the CPUs on which the
+        # process will be scheduled. It is represented as a bitmask, with the
+        # lowest order bit corresponding to the first logical CPU and the
+        # highest order bit corresponding to the last logical CPU. Not all CPUs
+        # may exist on a given system but a mask may specify more CPUs than are
+        # present.
+        #
+        # For example:
+        #    0x00000001  is processor #0,
+        #    0x00000003  is processors #0 and #1,
+        #    0xFFFFFFFF  is all processors (#0 through #31).
+        #
+        # Pinning a Python process to a single CPU is desirable, because Python
+        # is inherently single-threaded due to the GIL, and can suffer a
+        # 30-40%% slowdown due to cache blow-out and thread context switching
+        # if the scheduler happens to schedule the underlying threads across
+        # different cores. See
+        # https://www.mirantis.com/blog/improve-performance-python-programs-restricting-single-cpu/.
+        #
+        # cpu_affinity: 0xFFFFFFFF
+
        # Whether to serve a web client from the HTTP/HTTPS root resource.
        web_client: True

@@ -32,6 +32,9 @@ class WorkerConfig(Config):
        self.worker_replication_port = config.get("worker_replication_port", None)
        self.worker_name = config.get("worker_name", self.worker_app)

+        self.worker_main_http_uri = config.get("worker_main_http_uri", None)
+        self.worker_cpu_affinity = config.get("worker_cpu_affinity")
+
        if self.worker_listeners:
            for listener in self.worker_listeners:
                bind_address = listener.pop("bind_address", None)
@@ -435,6 +435,12 @@ class TransactionQueue(object):

                # END CRITICAL SECTION

+                if len(pending_pdus) > 10:
+                    logger.info("Sending %d pdus to %r", len(pending_pdus), destination)
+
+                if len(pending_edus) > 10:
+                    logger.info("Sending %d edus to %r", len(pending_edus), destination)
+
                success = yield self._send_new_transaction(
                    destination, pending_pdus, pending_edus, pending_failures,
                )
@@ -494,6 +500,12 @@ class TransactionQueue(object):
            for content in contents
        ]

+        if len(contents) > 10:
+            logger.info(
+                "Sending %d direct to device messages to %r",
+                len(contents), destination,
+            )
+
        last_device_list = self.last_device_list_stream_id_by_dest.get(destination, 0)
        now_stream_id, results = yield self.store.get_devices_by_remote(
            destination, last_device_list
@@ -507,6 +519,10 @@ class TransactionQueue(object):
            )
            for content in results
        )
+
+        if len(results) > 10:
+            logger.info("Sending %d device list updates to %r", len(results), destination)
+
        defer.returnValue((edus, stream_id, now_stream_id))

    @measure_func("_send_new_transaction")
@@ -153,12 +153,10 @@ class Authenticator(object):
 class BaseFederationServlet(object):
    REQUIRE_AUTH = True

-    def __init__(self, handler, authenticator, ratelimiter, server_name,
-                 room_list_handler):
+    def __init__(self, handler, authenticator, ratelimiter, server_name):
        self.handler = handler
        self.authenticator = authenticator
        self.ratelimiter = ratelimiter
-        self.room_list_handler = room_list_handler

    def _wrap(self, func):
        authenticator = self.authenticator
@@ -590,7 +588,7 @@ class PublicRoomList(BaseFederationServlet):
        else:
            network_tuple = ThirdPartyInstanceID(None, None)

-        data = yield self.room_list_handler.get_local_public_room_list(
+        data = yield self.handler.get_local_public_room_list(
            limit, since_token,
            network_tuple=network_tuple
        )
@@ -611,7 +609,7 @@ class FederationVersionServlet(BaseFederationServlet):
        }))


-SERVLET_CLASSES = (
+FEDERATION_SERVLET_CLASSES = (
    FederationSendServlet,
    FederationPullServlet,
    FederationEventServlet,
@@ -634,17 +632,27 @@ SERVLET_CLASSES = (
    FederationThirdPartyInviteExchangeServlet,
    On3pidBindServlet,
    OpenIdUserInfo,
-    PublicRoomList,
    FederationVersionServlet,
 )

+ROOM_LIST_CLASSES = (
+    PublicRoomList,
+)
+

 def register_servlets(hs, resource, authenticator, ratelimiter):
-    for servletclass in SERVLET_CLASSES:
+    for servletclass in FEDERATION_SERVLET_CLASSES:
        servletclass(
            handler=hs.get_replication_layer(),
            authenticator=authenticator,
            ratelimiter=ratelimiter,
            server_name=hs.hostname,
-            room_list_handler=hs.get_room_list_handler(),
+        ).register(resource)
+
+    for servletclass in ROOM_LIST_CLASSES:
+        servletclass(
+            handler=hs.get_room_list_handler(),
+            authenticator=authenticator,
+            ratelimiter=ratelimiter,
+            server_name=hs.hostname,
        ).register(resource)
@@ -76,6 +76,7 @@ class FederationHandler(BaseHandler):
        self.keyring = hs.get_keyring()
        self.action_generator = hs.get_action_generator()
        self.is_mine_id = hs.is_mine_id
+        self.pusher_pool = hs.get_pusherpool()

        self.replication_layer.set_handler(self)

@@ -1412,7 +1413,7 @@ class FederationHandler(BaseHandler):
            auth_events=auth_events,
        )

-        if not event.internal_metadata.is_outlier():
+        if not event.internal_metadata.is_outlier() and not backfilled:
            yield self.action_generator.handle_push_actions_for_event(
                event, context
            )
@@ -1426,7 +1427,7 @@ class FederationHandler(BaseHandler):
        if not backfilled:
            # this intentionally does not yield: we don't care about the result
            # and don't need to wait for it.
-            preserve_fn(self.hs.get_pusherpool().on_new_notifications)(
+            preserve_fn(self.pusher_pool.on_new_notifications)(
                event_stream_id, max_stream_id
            )

@@ -50,6 +50,8 @@ class MessageHandler(BaseHandler):

        self.pagination_lock = ReadWriteLock()

+        self.pusher_pool = hs.get_pusherpool()
+
        # We arbitrarily limit concurrent event creation for a room to 5.
        # This is to stop us from diverging history *too* much.
        self.limiter = Limiter(max_count=5)
@@ -610,7 +612,7 @@ class MessageHandler(BaseHandler):

        # this intentionally does not yield: we don't care about the result
        # and don't need to wait for it.
-        preserve_fn(self.hs.get_pusherpool().on_new_notifications)(
+        preserve_fn(self.pusher_pool.on_new_notifications)(
            event_stream_id, max_stream_id
        )

@@ -579,18 +579,17 @@ class SyncHandler(object):
        since_token = sync_result_builder.since_token

        if since_token and since_token.device_list_key:
-            room_ids = yield self.store.get_rooms_for_user(user_id)
-
-            user_ids_changed = set()
            changed = yield self.store.get_user_whose_devices_changed(
                since_token.device_list_key
            )
-            for other_user_id in changed:
-                other_room_ids = yield self.store.get_rooms_for_user(other_user_id)
-                if room_ids.intersection(other_room_ids):
-                    user_ids_changed.add(other_user_id)
+            if not changed:
+                defer.returnValue([])

-            defer.returnValue(user_ids_changed)
+            users_who_share_room = yield self.store.get_users_who_share_room_with_user(
+                user_id
+            )
+
+            defer.returnValue(users_who_share_room & changed)
        else:
            defer.returnValue([])

@@ -19,8 +19,9 @@ from twisted.internet import defer

 from .push_rule_evaluator import PushRuleEvaluatorForEvent

-from synapse.visibility import filter_events_for_clients_context
 from synapse.api.constants import EventTypes, Membership
+from synapse.metrics import get_metrics_for
+from synapse.util.caches import metrics as cache_metrics
 from synapse.util.caches.descriptors import cached
 from synapse.util.async import Linearizer

@@ -32,6 +33,23 @@ logger = logging.getLogger(__name__)

 rules_by_room = {}

+push_metrics = get_metrics_for(__name__)
+
+push_rules_invalidation_counter = push_metrics.register_counter(
+    "push_rules_invalidation_counter"
+)
+push_rules_state_size_counter = push_metrics.register_counter(
+    "push_rules_state_size_counter"
+)
+
+# Measures whether we use the fast path of using state deltas, or if we have to
+# recalculate from scratch
+push_rules_delta_state_cache_metric = cache_metrics.register_cache(
+    "cache",
+    size_callback=lambda: 0,  # Meaningless size, as this isn't a cache that stores values
+    cache_name="push_rules_delta_state_cache_metric",
+)
+

 class BulkPushRuleEvaluator(object):
    """Calculates the outcome of push rules for an event for all users in the
@@ -42,6 +60,12 @@ class BulkPushRuleEvaluator(object):
        self.hs = hs
        self.store = hs.get_datastore()

+        self.room_push_rule_cache_metrics = cache_metrics.register_cache(
+            "cache",
+            size_callback=lambda: 0,  # There's not good value for this
+            cache_name="room_push_rule_cache",
+        )
+
    @defer.inlineCallbacks
    def _get_rules_for_event(self, event, context):
        """This gets the rules for all users in the room at the time of the event,
@@ -79,7 +103,10 @@ class BulkPushRuleEvaluator(object):
        # It's important that RulesForRoom gets added to self._get_rules_for_room.cache
        # before any lookup methods get called on it as otherwise there may be
        # a race if invalidate_all gets called (which assumes its in the cache)
-        return RulesForRoom(self.hs, room_id, self._get_rules_for_room.cache)
+        return RulesForRoom(
+            self.hs, room_id, self._get_rules_for_room.cache,
+            self.room_push_rule_cache_metrics,
+        )

    @defer.inlineCallbacks
    def action_for_event_by_user(self, event, context):
@@ -92,15 +119,6 @@ class BulkPushRuleEvaluator(object):
        rules_by_user = yield self._get_rules_for_event(event, context)
        actions_by_user = {}

-        # None of these users can be peeking since this list of users comes
-        # from the set of users in the room, so we know for sure they're all
-        # actually in the room.
-        user_tuples = [(u, False) for u in rules_by_user]
-
-        filtered_by_user = yield filter_events_for_clients_context(
-            self.store, user_tuples, [event], {event.event_id: context}
-        )
-
        room_members = yield self.store.get_joined_users_from_context(
            event, context
        )
@@ -110,6 +128,14 @@ class BulkPushRuleEvaluator(object):
        condition_cache = {}

        for uid, rules in rules_by_user.iteritems():
+            if event.sender == uid:
+                continue
+
+            if not event.is_state():
+                is_ignored = yield self.store.is_ignored_by(event.sender, uid)
+                if is_ignored:
+                    continue
+
            display_name = None
            profile_info = room_members.get(uid)
            if profile_info:
@@ -121,13 +147,6 @@ class BulkPushRuleEvaluator(object):
                if event.type == EventTypes.Member and event.state_key == uid:
                    display_name = event.content.get("displayname", None)

-            filtered = filtered_by_user[uid]
-            if len(filtered) == 0:
-                continue
-
-            if filtered[0].sender == uid:
-                continue
-
            for rule in rules:
                if 'enabled' in rule and not rule['enabled']:
                    continue
@@ -170,17 +189,19 @@ class RulesForRoom(object):
    the entire cache for the room.
    """

-    def __init__(self, hs, room_id, rules_for_room_cache):
+    def __init__(self, hs, room_id, rules_for_room_cache, room_push_rule_cache_metrics):
        """
        Args:
            hs (HomeServer)
            room_id (str)
            rules_for_room_cache(Cache): The cache object that caches these
                RoomsForUser objects.
+            room_push_rule_cache_metrics (CacheMetric)
        """
        self.room_id = room_id
        self.is_mine_id = hs.is_mine_id
        self.store = hs.get_datastore()
+        self.room_push_rule_cache_metrics = room_push_rule_cache_metrics

        self.linearizer = Linearizer(name="rules_for_room")

@@ -222,11 +243,19 @@ class RulesForRoom(object):
        """
        state_group = context.state_group

+        if state_group and self.state_group == state_group:
+            logger.debug("Using cached rules for %r", self.room_id)
+            self.room_push_rule_cache_metrics.inc_hits()
+            defer.returnValue(self.rules_by_user)
+
        with (yield self.linearizer.queue(())):
            if state_group and self.state_group == state_group:
                logger.debug("Using cached rules for %r", self.room_id)
+                self.room_push_rule_cache_metrics.inc_hits()
                defer.returnValue(self.rules_by_user)

+            self.room_push_rule_cache_metrics.inc_misses()
+
            ret_rules_by_user = {}
            missing_member_event_ids = {}
            if state_group and self.state_group == context.prev_group:
@@ -234,8 +263,13 @@ class RulesForRoom(object):
                # results.
                ret_rules_by_user = self.rules_by_user
                current_state_ids = context.delta_ids
+
+                push_rules_delta_state_cache_metric.inc_hits()
            else:
                current_state_ids = context.current_state_ids
+                push_rules_delta_state_cache_metric.inc_misses()
+
+            push_rules_state_size_counter.inc_by(len(current_state_ids))

            logger.debug(
                "Looking for member changes in %r %r", state_group, current_state_ids
@@ -282,6 +316,14 @@ class RulesForRoom(object):
                yield self._update_rules_with_member_event_ids(
                    ret_rules_by_user, missing_member_event_ids, state_group, event
                )
+            else:
+                # The push rules didn't change but lets update the cache anyway
+                self.update_cache(
+                    self.sequence,
+                    members={},  # There were no membership changes
+                    rules_by_user=ret_rules_by_user,
+                    state_group=state_group
+                )

        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
@@ -380,6 +422,7 @@ class RulesForRoom(object):
        self.state_group = object()
        self.member_map = {}
        self.rules_by_user = {}
+        push_rules_invalidation_counter.inc()

    def update_cache(self, sequence, members, rules_by_user, state_group):
        if sequence == self.sequence:
@@ -200,7 +200,9 @@ def _glob_to_re(glob, word_boundary):
        return re.compile(r, flags=re.IGNORECASE)


-def _flatten_dict(d, prefix=[], result={}):
+def _flatten_dict(d, prefix=[], result=None):
+    if result is None:
+        result = {}
    for key, value in d.items():
        if isinstance(value, basestring):
            result[".".join(prefix + [key])] = value.lower()
@@ -31,7 +31,7 @@ REQUIREMENTS = {
    "pyyaml": ["yaml"],
    "pyasn1": ["pyasn1"],
    "daemonize": ["daemonize"],
-    "py-bcrypt": ["bcrypt"],
+    "bcrypt": ["bcrypt"],
    "pillow": ["PIL"],
    "pydenticon": ["pydenticon"],
    "ujson": ["ujson"],
@@ -40,6 +40,7 @@ REQUIREMENTS = {
    "pymacaroons-pynacl": ["pymacaroons"],
    "msgpack-python>=0.3.0": ["msgpack"],
    "phonenumbers>=8.2.0": ["phonenumbers"],
+    "affinity": ["affinity"],
 }
 CONDITIONAL_REQUIREMENTS = {
    "web_client": {
@@ -29,9 +29,8 @@ class SlavedClientIpStore(BaseSlavedStore):
            max_entries=50000 * CACHE_SIZE_FACTOR,
        )

-    def insert_client_ip(self, user, access_token, ip, user_agent, device_id):
+    def insert_client_ip(self, user_id, access_token, ip, user_agent, device_id):
        now = int(self._clock.time_msec())
-        user_id = user.to_string()
        key = (user_id, access_token, ip)

        try:
@@ -323,14 +323,18 @@ class UserIpCommand(Command):

    @classmethod
    def from_line(cls, line):
-        user_id, access_token, ip, device_id, last_seen, user_agent = line.split(" ", 5)
+        user_id, jsn = line.split(" ", 1)

-        return cls(user_id, access_token, ip, user_agent, device_id, int(last_seen))
+        access_token, ip, user_agent, device_id, last_seen = json.loads(jsn)
+
+        return cls(
+            user_id, access_token, ip, user_agent, device_id, last_seen
+        )

    def to_line(self):
-        return " ".join((
-            self.user_id, self.access_token, self.ip, self.device_id,
-            str(self.last_seen), self.user_agent,
+        return self.user_id + " " + json.dumps((
+            self.access_token, self.ip, self.user_agent, self.device_id,
+            self.last_seen,
        ))


@@ -244,7 +244,7 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
                becoming full.
        """
        if self.state == ConnectionStates.CLOSED:
-            logger.info("[%s] Not sending, connection closed", self.id())
+            logger.debug("[%s] Not sending, connection closed", self.id())
            return

        if do_buffer and self.state != ConnectionStates.ESTABLISHED:
@@ -264,7 +264,7 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
    def _queue_command(self, cmd):
        """Queue the command until the connection is ready to write to again.
        """
-        logger.info("[%s] Queing as conn %r, cmd: %r", self.id(), self.state, cmd)
+        logger.debug("[%s] Queing as conn %r, cmd: %r", self.id(), self.state, cmd)
        self.pending_commands.append(cmd)

        if len(self.pending_commands) > self.max_line_buffer:
@@ -168,7 +168,7 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):

    DEFAULT_MESSAGE = (
        "Sharing illegal content on this server is not permitted and rooms in"
-        " violatation will be blocked."
+        " violation will be blocked."
    )

    def __init__(self, hs):
@@ -296,7 +296,7 @@ class QuarantineMediaInRoom(ClientV1RestServlet):

 class ResetPasswordRestServlet(ClientV1RestServlet):
    """Post request to allow an administrator reset password for a user.
-    This need a user have a administrator access in Synapse.
+    This needs user to have administrator access in Synapse.
        Example:
            http://localhost:8008/_matrix/client/api/v1/admin/reset_password/
            @user:to_reset_password?access_token=admin_access_token
@@ -319,7 +319,7 @@ class ResetPasswordRestServlet(ClientV1RestServlet):
    @defer.inlineCallbacks
    def on_POST(self, request, target_user_id):
        """Post request to allow an administrator reset password for a user.
-        This need a user have a administrator access in Synapse.
+        This needs user to have administrator access in Synapse.
        """
        UserID.from_string(target_user_id)
        requester = yield self.auth.get_user_by_req(request)
@@ -343,7 +343,7 @@ class ResetPasswordRestServlet(ClientV1RestServlet):

 class GetUsersPaginatedRestServlet(ClientV1RestServlet):
    """Get request to get specific number of users from Synapse.
-    This need a user have a administrator access in Synapse.
+    This needs user to have administrator access in Synapse.
        Example:
            http://localhost:8008/_matrix/client/api/v1/admin/users_paginate/
            @admin:user?access_token=admin_access_token&start=0&limit=10
@@ -362,7 +362,7 @@ class GetUsersPaginatedRestServlet(ClientV1RestServlet):
    @defer.inlineCallbacks
    def on_GET(self, request, target_user_id):
        """Get request to get specific number of users from Synapse.
-        This need a user have a administrator access in Synapse.
+        This needs user to have administrator access in Synapse.
        """
        target_user = UserID.from_string(target_user_id)
        requester = yield self.auth.get_user_by_req(request)
@@ -395,7 +395,7 @@ class GetUsersPaginatedRestServlet(ClientV1RestServlet):
    @defer.inlineCallbacks
    def on_POST(self, request, target_user_id):
        """Post request to get specific number of users from Synapse..
-        This need a user have a administrator access in Synapse.
+        This needs user to have administrator access in Synapse.
        Example:
            http://localhost:8008/_matrix/client/api/v1/admin/users_paginate/
            @admin:user?access_token=admin_access_token
@@ -433,7 +433,7 @@ class GetUsersPaginatedRestServlet(ClientV1RestServlet):
 class SearchUsersRestServlet(ClientV1RestServlet):
    """Get request to search user table for specific users according to
    search term.
-    This need a user have a administrator access in Synapse.
+    This needs user to have administrator access in Synapse.
        Example:
            http://localhost:8008/_matrix/client/api/v1/admin/search_users/
            @admin:user?access_token=admin_access_token&term=alice
@@ -453,7 +453,7 @@ class SearchUsersRestServlet(ClientV1RestServlet):
    def on_GET(self, request, target_user_id):
        """Get request to search user table for specific users according to
        search term.
-        This need a user have a administrator access in Synapse.
+        This needs user to have a administrator access in Synapse.
        """
        target_user = UserID.from_string(target_user_id)
        requester = yield self.auth.get_user_by_req(request)
@@ -73,6 +73,7 @@ class PushersSetRestServlet(ClientV1RestServlet):
    def __init__(self, hs):
        super(PushersSetRestServlet, self).__init__(hs)
        self.notifier = hs.get_notifier()
+        self.pusher_pool = self.hs.get_pusherpool()

    @defer.inlineCallbacks
    def on_POST(self, request):
@@ -81,12 +82,10 @@ class PushersSetRestServlet(ClientV1RestServlet):

        content = parse_json_object_from_request(request)

-        pusher_pool = self.hs.get_pusherpool()
-
        if ('pushkey' in content and 'app_id' in content
                and 'kind' in content and
                content['kind'] is None):
-            yield pusher_pool.remove_pusher(
+            yield self.pusher_pool.remove_pusher(
                content['app_id'], content['pushkey'], user_id=user.to_string()
            )
            defer.returnValue((200, {}))
@@ -109,14 +108,14 @@ class PushersSetRestServlet(ClientV1RestServlet):
            append = content['append']

        if not append:
-            yield pusher_pool.remove_pushers_by_app_id_and_pushkey_not_user(
+            yield self.pusher_pool.remove_pushers_by_app_id_and_pushkey_not_user(
                app_id=content['app_id'],
                pushkey=content['pushkey'],
                not_user_id=user.to_string()
            )

        try:
-            yield pusher_pool.add_pusher(
+            yield self.pusher_pool.add_pusher(
                user_id=user.to_string(),
                access_token=requester.access_token_id,
                kind=content['kind'],
@@ -152,6 +151,7 @@ class PushersRemoveRestServlet(RestServlet):
        self.hs = hs
        self.notifier = hs.get_notifier()
        self.auth = hs.get_v1auth()
+        self.pusher_pool = self.hs.get_pusherpool()

    @defer.inlineCallbacks
    def on_GET(self, request):
@@ -161,10 +161,8 @@ class PushersRemoveRestServlet(RestServlet):
        app_id = parse_string(request, "app_id", required=True)
        pushkey = parse_string(request, "pushkey", required=True)

-        pusher_pool = self.hs.get_pusherpool()
-
        try:
-            yield pusher_pool.remove_pusher(
+            yield self.pusher_pool.remove_pusher(
                app_id=app_id,
                pushkey=pushkey,
                user_id=user.to_string(),
@@ -110,7 +110,7 @@ class SyncRestServlet(RestServlet):
        filter_id = parse_string(request, "filter", default=None)
        full_state = parse_boolean(request, "full_state", default=False)

-        logger.info(
+        logger.debug(
            "/sync: user=%r, timeout=%r, since=%r,"
            " set_presence=%r, filter_id=%r, device_id=%r" % (
                user, timeout, since, set_presence, filter_id, device_id
@@ -164,27 +164,34 @@ class SyncRestServlet(RestServlet):
            )

        time_now = self.clock.time_msec()
-
-        joined = self.encode_joined(
-            sync_result.joined, time_now, requester.access_token_id, filter.event_fields
+        response_content = self.encode_response(
+            time_now, sync_result, requester.access_token_id, filter
        )

-        invited = self.encode_invited(
-            sync_result.invited, time_now, requester.access_token_id
+        defer.returnValue((200, response_content))
+
+    @staticmethod
+    def encode_response(time_now, sync_result, access_token_id, filter):
+        joined = SyncRestServlet.encode_joined(
+            sync_result.joined, time_now, access_token_id, filter.event_fields
        )

-        archived = self.encode_archived(
-            sync_result.archived, time_now, requester.access_token_id,
+        invited = SyncRestServlet.encode_invited(
+            sync_result.invited, time_now, access_token_id,
+        )
+
+        archived = SyncRestServlet.encode_archived(
+            sync_result.archived, time_now, access_token_id,
            filter.event_fields,
        )

-        response_content = {
+        return {
            "account_data": {"events": sync_result.account_data},
            "to_device": {"events": sync_result.to_device},
            "device_lists": {
                "changed": list(sync_result.device_lists),
            },
-            "presence": self.encode_presence(
+            "presence": SyncRestServlet.encode_presence(
                sync_result.presence, time_now
            ),
            "rooms": {
@@ -196,9 +203,8 @@ class SyncRestServlet(RestServlet):
            "next_batch": sync_result.next_batch.to_string(),
        }

-        defer.returnValue((200, response_content))
-
-    def encode_presence(self, events, time_now):
+    @staticmethod
+    def encode_presence(events, time_now):
        return {
            "events": [
                {
@@ -212,7 +218,8 @@ class SyncRestServlet(RestServlet):
            ]
        }

-    def encode_joined(self, rooms, time_now, token_id, event_fields):
+    @staticmethod
+    def encode_joined(rooms, time_now, token_id, event_fields):
        """
        Encode the joined rooms in a sync result

@@ -231,13 +238,14 @@ class SyncRestServlet(RestServlet):
        """
        joined = {}
        for room in rooms:
-            joined[room.room_id] = self.encode_room(
+            joined[room.room_id] = SyncRestServlet.encode_room(
                room, time_now, token_id, only_fields=event_fields
            )

        return joined

-    def encode_invited(self, rooms, time_now, token_id):
+    @staticmethod
+    def encode_invited(rooms, time_now, token_id):
        """
        Encode the invited rooms in a sync result

@@ -270,7 +278,8 @@ class SyncRestServlet(RestServlet):

        return invited

-    def encode_archived(self, rooms, time_now, token_id, event_fields):
+    @staticmethod
+    def encode_archived(rooms, time_now, token_id, event_fields):
        """
        Encode the archived rooms in a sync result

@@ -289,7 +298,7 @@ class SyncRestServlet(RestServlet):
        """
        joined = {}
        for room in rooms:
-            joined[room.room_id] = self.encode_room(
+            joined[room.room_id] = SyncRestServlet.encode_room(
                room, time_now, token_id, joined=False, only_fields=event_fields
            )

@@ -308,3 +308,16 @@ class AccountDataStore(SQLBaseStore):
            " WHERE stream_id < ?"
        )
        txn.execute(update_max_id_sql, (next_id, next_id))
+
+    @cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)
+    def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
+        ignored_account_data = yield self.get_global_account_data_by_type_for_user(
+            "m.ignored_user_list", ignorer_user_id,
+            on_invalidate=cache_context.invalidate,
+        )
+        if not ignored_account_data:
+            defer.returnValue(False)
+
+        defer.returnValue(
+            ignored_user_id in ignored_account_data.get("ignored_users", {})
+        )
@@ -56,9 +56,11 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
        )
        reactor.addSystemEventTrigger("before", "shutdown", self._update_client_ips_batch)

-    def insert_client_ip(self, user, access_token, ip, user_agent, device_id):
-        now = int(self._clock.time_msec())
-        key = (user.to_string(), access_token, ip)
+    def insert_client_ip(self, user_id, access_token, ip, user_agent, device_id,
+                         now=None):
+        if not now:
+            now = int(self._clock.time_msec())
+        key = (user_id, access_token, ip)

        try:
            last_seen = self.client_ip_last_seen.get(key)
@@ -188,25 +188,6 @@ def filter_events_for_clients(store, user_tuples, events, event_id_to_state):
    })


-@defer.inlineCallbacks
-def filter_events_for_clients_context(store, user_tuples, events, event_id_to_context):
-    user_ids = set(u[0] for u in user_tuples)
-    event_id_to_state = {}
-    for event_id, context in event_id_to_context.items():
-        state = yield store.get_events([
-            e_id
-            for key, e_id in context.current_state_ids.iteritems()
-            if key == (EventTypes.RoomHistoryVisibility, "")
-            or (key[0] == EventTypes.Member and key[1] in user_ids)
-        ])
-        event_id_to_state[event_id] = state
-
-    res = yield filter_events_for_clients(
-        store, user_tuples, events, event_id_to_state
-    )
-    defer.returnValue(res)
-
-
@defer.inlineCallbacks
 def filter_events_for_client(store, user_id, events, is_peeking=False):
    """
@@ -19,7 +19,6 @@ import synapse.api.errors
 import synapse.handlers.device

 import synapse.storage
-from synapse import types
 from tests import unittest, utils

 user1 = "@boris:aaa"
@@ -179,6 +178,6 @@ class DeviceTestCase(unittest.TestCase):

        if ip is not None:
            yield self.store.insert_client_ip(
-                types.UserID.from_string(user_id),
+                user_id,
                access_token, ip, "user_agent", device_id)
            self.clock.advance_time(1000)
@@ -15,9 +15,6 @@

 from twisted.internet import defer

-import synapse.server
-import synapse.storage
-import synapse.types
 import tests.unittest
 import tests.utils

@@ -39,7 +36,7 @@ class ClientIpStoreTestCase(tests.unittest.TestCase):
        self.clock.now = 12345678
        user_id = "@user:id"
        yield self.store.insert_client_ip(
-            synapse.types.UserID.from_string(user_id),
+            user_id,
            "access_token", "ip", "user_agent", "device_id",
        )

@@ -14,14 +14,38 @@ deps =

 setenv =
    PYTHONDONTWRITEBYTECODE = no_byte_code
-    # As of twisted 16.4, trial tries to import the tests as a package, which
-    # means it needs to be on the pythonpath.
-    PYTHONPATH = {toxinidir}
+
 commands =
-    /bin/sh -c "find {toxinidir} -name '*.pyc' -delete ; coverage run {env:COVERAGE_OPTS:} --source={toxinidir}/synapse \
-        {envbindir}/trial {env:TRIAL_FLAGS:} {posargs:tests} {env:TOXSUFFIX:}"
+    /usr/bin/find "{toxinidir}" -name '*.pyc' -delete
+    coverage run {env:COVERAGE_OPTS:} --source="{toxinidir}/synapse" \
+        "{envbindir}/trial" {env:TRIAL_FLAGS:} {posargs:tests} {env:TOXSUFFIX:}
    {env:DUMP_COVERAGE_COMMAND:coverage report -m}

+[testenv:py27]
+
+# As of twisted 16.4, trial tries to import the tests as a package (previously
+# it loaded the files explicitly), which means they need to be on the
+# pythonpath. Our sdist doesn't include the 'tests' package, so normally it
+# doesn't work within the tox virtualenv.
+#
+# As a workaround, we tell tox to do install with 'pip -e', which just
+# creates a symlink to the project directory instead of unpacking the sdist.
+#
+# (An alternative to this would be to set PYTHONPATH to include the project
+# directory. Note two problems with this:
+#
+#   - if you set it via `setenv`, then it is also set during the 'install'
+#     phase, which inhibits unpacking the sdist, so the virtualenv isn't
+#     useful for anything else without setting PYTHONPATH similarly.
+#
+#   - `synapse` is also loaded from PYTHONPATH so even if you only set
+#     PYTHONPATH for the test phase, we're still running the tests against
+#     the working copy rather than the contents of the sdist. So frankly
+#     you might as well use -e in the first place.
+#
+# )
+usedevelop=true
+
 [testenv:packaging]
 deps =
    check-manifest
Author	SHA1	Message	Date
Erik Johnston	a991cf036e	Add logging for many pdu/edus	2017-09-11 16:28:20 +01:00
Richard van der Hoff	e2fcba038c	Merge pull request #2439 from matrix-org/rav/tox_tweaks do tox install with pip -e	2017-09-06 17:08:19 +01:00
Richard van der Hoff	5f59f20636	Merge remote-tracking branch 'origin/master' into develop	2017-09-05 21:58:19 +01:00
Richard van der Hoff	59de2c7afa	Exclude the github issue template from our sdist (#2440 ) PR #2413 added an issue template, but just adding files to the project directory upsets the packaging scripts: we need to explicitly include or exclude them. Move the template into a .github directory to make that easy, and to de-clutter the root a bit.	2017-09-05 21:57:19 +01:00
Richard van der Hoff	4b616c8cf2	Merge branch 'master' into develop	2017-09-05 17:51:13 +01:00
Richard van der Hoff	4dd61df6f8	do tox install with pip -e - this ensures we end up with a working virtualenv which we can use for other things.	2017-09-05 16:35:23 +01:00
Erik Johnston	c0c31656ff	Merge pull request #2433 from ptman/patch-1 Document known to work postgres version	2017-09-01 15:28:42 +01:00
Paul Tötterman	8b16b43b7f	Document known to work postgres version	2017-09-01 16:52:45 +03:00
Richard van der Hoff	dff396de0f	Set --python when running sytest .. because I want to make the 'install_and_run' script useful for non-synapse jobs, which do not accept --python. In any case we set up the path here, so sytest shouldn't be guessing it.	2017-09-01 11:20:37 +01:00
Richard van der Hoff	f06ffdb6fa	fix python path in jenkins scripts	2017-09-01 10:31:45 +01:00
Richard van der Hoff	6e67aaa7f2	Set --python when running sytest .. because I want to make the 'install_and_run' script useful for non-synapse jobs, which do not accept --python. In any case we set up the path here, so sytest shouldn't be guessing it.	2017-09-01 10:06:21 +01:00
Richard van der Hoff	934ab76835	Merge pull request #2428 from matrix-org/rav/update_upgrade Tweaks to the upgrade instructions	2017-08-24 11:42:02 +01:00
Richard van der Hoff	fc9878f6a4	Tweaks to the upgrade instructions	2017-08-23 15:27:02 +01:00
Richard van der Hoff	a4d3bfe3d6	Merge pull request #2417 from matrix-org/rav/federation_client Improvements to the federation test client	2017-08-23 14:50:26 +01:00
Richard van der Hoff	a7effa8400	Merge pull request #2288 from kyrias/bcrypt python_dependencies: Use bcrypt module instead of py-bcrypt	2017-08-23 14:14:56 +01:00
Richard van der Hoff	a04c6bbf8f	test federation client: Allow server-name and key-file as options so that you don't necessarily need a config file.	2017-08-22 11:19:30 +01:00
Richard van der Hoff	77ea8cbdd7	Merge pull request #2416 from matrix-org/rav/prometheus_config Add prometheus config	2017-08-22 10:34:40 +01:00
Tom Lant	20b3660495	Merge pull request #2413 from matrix-org/toml-issue-template Issue template for Synapse	2017-08-21 16:07:35 +01:00
Richard van der Hoff	046b659ce2	Improvements to the federation test client Make it read the config file, primarily.	2017-08-17 16:59:11 +01:00
Tom Lant	413c270723	Update ISSUE_TEMPLATE.md Added instructions for checking server version.	2017-08-17 11:14:35 +01:00
Tom Lant	ec3a2dc773	Update ISSUE_TEMPLATE.md Responding to review comments.	2017-08-17 11:00:51 +01:00
Richard van der Hoff	012875258c	Add prometheus config ... from https://github.com/matrix-org/synapse-prometheus-config.	2017-08-16 15:31:44 +01:00
Richard van der Hoff	692250c6be	Fix user_dir startup Add missing parameter to _base.start_worker_reactor	2017-08-16 15:11:29 +01:00
Richard van der Hoff	d2352347cf	Fix process startup escape the % that got added in `92168cb` so that the process starts up ok.	2017-08-16 14:57:35 +01:00
Matthew Hodgson	92168cbbc5	explain why CPU affinity is a good idea	2017-08-15 18:27:42 +01:00
Richard van der Hoff	963015005e	Merge pull request #2415 from matrix-org/rav/synctl_cpu_affinity Allow configuration of CPU affinity	2017-08-15 17:42:05 +01:00
Richard van der Hoff	10d8b701a1	Allow configuration of CPU affinity Make it possible to set the CPU affinity in the config file, so that we don't need to remember to do it manually every time.	2017-08-15 17:08:28 +01:00
Richard van der Hoff	543c794a76	Factor out common application start We have 10 copies of this code, and I don't really want to update each one separately.	2017-08-15 17:04:40 +01:00
Tom Lant	57cd0c3dea	Update ISSUE_TEMPLATE.md Removed the sentence encouraging people not to file a bug - if people are in doubt we'd rather they filed a bug than gave up entirely.	2017-08-14 14:40:32 +01:00
Tom Lant	b524dd4c35	Update ISSUE_TEMPLATE.md Oops capital L.	2017-08-14 14:36:49 +01:00
Tom Lant	09703609fc	Create ISSUE_TEMPLATE.md A new issue template proposed to try and steer people towards #matrix:matrix.org for support queries relating to running their own homeserver.	2017-08-14 14:35:25 +01:00
hera	eae04f1952	fix english	2017-08-04 23:56:42 +01:00
hera	5699b05072	typo	2017-08-04 23:44:37 +01:00
Erik Johnston	09552f9d9c	Reduce spammy log line in synchrotrons	2017-08-02 17:29:51 +01:00
Erik Johnston	b27429729d	Merge pull request #2375 from matrix-org/erikj/port_script Fix port script for user directory tables	2017-07-20 16:16:43 +01:00
Erik Johnston	60a9a49f83	Extend comment	2017-07-20 16:16:29 +01:00
Erik Johnston	d7d24750be	Fix port script for user directory tables	2017-07-20 10:47:01 +01:00
Erik Johnston	514c2d3c4d	Merge pull request #2371 from matrix-org/erikj/push_cache_hit Increase cache hit ratio for push	2017-07-17 09:42:27 +01:00
Erik Johnston	bfde076022	Increase cache hit ratio for push We don't update the cache in all code paths, which causes subsequent calls to miss the cache	2017-07-14 16:11:26 +01:00
Erik Johnston	d3862812ff	Merge pull request #2366 from matrix-org/erikj/push_metrics Add more metrics to push rule evaluation	2017-07-14 11:04:03 +01:00
Erik Johnston	8d26385d76	Add more metrics to push rule evaluation	2017-07-13 14:37:30 +01:00
Erik Johnston	67b7b904ba	Merge pull request #2365 from matrix-org/erikj/push_skip_lock Push: Don't acquire lock unless necessary	2017-07-13 11:44:48 +01:00
Erik Johnston	f60218ec41	Push: Don't acquire lock unless necessary	2017-07-13 11:23:53 +01:00
Erik Johnston	91818723a1	Merge pull request #2362 from matrix-org/erikj/sync_user_users_who_share Use less DB for device list handling in sync	2017-07-12 10:45:30 +01:00
Erik Johnston	e9aec001f4	Use less DB for device list handling in sync	2017-07-12 10:30:10 +01:00
Erik Johnston	0184a97dbd	Merge pull request #2354 from krombel/reduce_static_sync_reply encode sync-response statically	2017-07-11 14:19:56 +01:00
Krombel	85b9f76f1d	split out reducing stuff; just make encode_* static	2017-07-11 13:14:35 +02:00
Erik Johnston	e2cb760dcc	Merge pull request #2357 from matrix-org/erikj/push Don't compute push actions for backfilled events	2017-07-11 10:53:22 +01:00
Erik Johnston	925b3638ff	Reduce log levels in tcp replication	2017-07-11 10:04:21 +01:00
Erik Johnston	9a6fd3ef29	Don't compute push actions for backfilled events	2017-07-11 10:02:21 +01:00
Krombel	2f82de18ee	fix test	2017-07-10 17:34:58 +02:00
Krombel	6e16aca8b0	encode sync-response statically; omit empty objects from sync-response	2017-07-10 16:42:17 +02:00
Erik Johnston	d4d12daed9	Include registration and as stores in frontend proxy	2017-07-07 18:36:45 +01:00
Erik Johnston	f467a8f66d	Merge branch 'master' of github.com:matrix-org/synapse into develop	2017-07-07 18:26:28 +01:00
Erik Johnston	c9184ed87e	Merge pull request #2344 from matrix-org/erikj/frontend_proxy Add a frontend proxy	2017-07-07 18:25:46 +01:00
Erik Johnston	1fc4a962e4	Add a frontend proxy	2017-07-07 18:19:46 +01:00
Erik Johnston	08284c86ed	Merge pull request #2343 from matrix-org/erikj/fastpush Perf: Don't filter events for push	2017-07-07 14:25:42 +01:00
Erik Johnston	f502b0dea1	Perf: Don't filter events for push We know the users are joined and we can explicitly check for if they are ignoring the user, so lets do that.	2017-07-07 14:04:40 +01:00
Erik Johnston	1200f28d66	Merge branch 'hotfixes-v0.22.1' of github.com:matrix-org/synapse	2017-07-06 18:11:49 +01:00
Erik Johnston	76ed3476d3	Bump version and changelog	2017-07-06 18:11:22 +01:00
Erik Johnston	58dc1f2c78	Merge pull request #2342 from matrix-org/erikj/pusher_pool_instantiate Fix bug where pusherpool didn't start and broke some rooms	2017-07-06 18:08:43 +01:00
Erik Johnston	5a7f561a9b	Fix bug where pusherpool didn't start and broke some rooms Since we didn't instansiate the PusherPool at start time it could fail at run time, which it did for some users. This may or may not fix things for those users, but it should happen at start time and stop the server from starting.	2017-07-06 17:55:51 +01:00
Erik Johnston	ed9a7f5436	Merge pull request #2309 from matrix-org/erikj/user_ip_repl Fix up user_ip replication commands	2017-07-06 14:33:14 +01:00
Erik Johnston	1f64207f26	Merge branch 'master' of github.com:matrix-org/synapse into develop	2017-07-06 13:57:45 +01:00
Erik Johnston	42b50483be	Merge branch 'release-v0.22.0' of github.com:matrix-org/synapse	2017-07-06 10:36:25 +01:00
Erik Johnston	6264cf9666	Bump version and changelog	2017-07-06 10:35:56 +01:00
Erik Johnston	f386632800	Merge pull request #2334 from matrix-org/erikj/refactor_transport_server Separate federation servlet into different lists	2017-07-05 17:09:07 +01:00
Erik Johnston	5e49a57ecc	Separate federation servlet into different lists	2017-07-05 14:32:24 +01:00
Richard van der Hoff	3d31b39297	Merge pull request #2332 from matrix-org/rav/fix_pushes Fix caching error in the push evaluator	2017-07-05 11:10:53 +01:00
Richard van der Hoff	73cfe48031	Fix caching error in the push evaluator Initialising `result` to `{}` in the parameters meant that every call to _flatten_dict used the same target dictionary. I'm hopeful this will fix https://github.com/matrix-org/synapse/issues/2270, but I suspect it won't. (This code seems to have been here since forever, unlike the bug, and I don't really think it explains the observed behaviour). Still, it makes it hard to investigate the problem.	2017-07-05 00:28:43 +01:00
Erik Johnston	27f26e48b7	Serialize user ip command as json	2017-06-27 16:25:38 +01:00
Erik Johnston	8c23221666	Fix up	2017-06-27 15:53:45 +01:00
Johannes Löthberg	4f66312df8	python_dependencies: Use bcrypt module instead of py-bcrypt py-bcrypt has been unmaintained for a long while, while bcrypt is actively maintained. And since `ff8b87118d` we're compatible with the bcrypt anyway. Signed-off-by: Johannes Löthberg <johannes@kyriasis.com>	2017-06-17 17:39:35 +02:00