mirror of https://gitlab.com/bashrc2/epicyon
Use a ring buffer for outbox threads
parent
343c9771bb
commit
2f0f48c346
106
daemon.py
106
daemon.py
|
|
@ -296,8 +296,8 @@ from utils import dangerousMarkup
|
|||
from utils import refreshNewswire
|
||||
from utils import isImageFile
|
||||
from utils import hasGroupType
|
||||
from manualapprove import manualDenyFollowRequest
|
||||
from manualapprove import manualApproveFollowRequest
|
||||
from manualapprove import manualDenyFollowRequestThread
|
||||
from manualapprove import manualApproveFollowRequestThread
|
||||
from announce import createAnnounce
|
||||
from content import getPriceFromString
|
||||
from content import replaceEmojiFromTags
|
||||
|
|
@ -1235,6 +1235,37 @@ class PubServer(BaseHTTPRequestHandler):
|
|||
self.server.CWlists,
|
||||
self.server.listsEnabled)
|
||||
|
||||
def _getOutboxThreadIndex(self, nickname: str,
|
||||
maxOutboxThreadsPerAccount: int) -> int:
|
||||
"""Returns the outbox thread index for the given account
|
||||
This is a ring buffer used to store the thread objects which
|
||||
are sending out posts
|
||||
"""
|
||||
accountOutboxThreadName = nickname
|
||||
if not accountOutboxThreadName:
|
||||
accountOutboxThreadName = '*'
|
||||
|
||||
# create the buffer for the given account
|
||||
if not self.server.outboxThread.get(accountOutboxThreadName):
|
||||
self.server.outboxThread[accountOutboxThreadName] = \
|
||||
[None] * maxOutboxThreadsPerAccount
|
||||
self.server.outboxThreadIndex[accountOutboxThreadName] = 0
|
||||
return 0
|
||||
|
||||
# increment the ring buffer index
|
||||
index = self.server.outboxThreadIndex[accountOutboxThreadName] + 1
|
||||
if index >= maxOutboxThreadsPerAccount:
|
||||
index = 0
|
||||
|
||||
self.server.outboxThreadIndex[accountOutboxThreadName] = index
|
||||
|
||||
# remove any existing thread from the current index in the buffer
|
||||
if self.server.outboxThread.get(accountOutboxThreadName):
|
||||
acct = accountOutboxThreadName
|
||||
if self.server.outboxThread[acct][index].is_alive():
|
||||
self.server.outboxThread[acct][index].kill()
|
||||
return index
|
||||
|
||||
def _postToOutboxThread(self, messageJson: {}) -> bool:
|
||||
"""Creates a thread to send a post
|
||||
"""
|
||||
|
|
@ -1242,24 +1273,18 @@ class PubServer(BaseHTTPRequestHandler):
|
|||
if not accountOutboxThreadName:
|
||||
accountOutboxThreadName = '*'
|
||||
|
||||
if self.server.outboxThread.get(accountOutboxThreadName):
|
||||
print('Waiting for previous outbox thread to end')
|
||||
waitCtr = 0
|
||||
thName = accountOutboxThreadName
|
||||
while self.server.outboxThread[thName].is_alive() and waitCtr < 8:
|
||||
time.sleep(1)
|
||||
waitCtr += 1
|
||||
if waitCtr >= 8:
|
||||
self.server.outboxThread[accountOutboxThreadName].kill()
|
||||
index = self._getOutboxThreadIndex(accountOutboxThreadName, 8)
|
||||
|
||||
print('Creating outbox thread')
|
||||
self.server.outboxThread[accountOutboxThreadName] = \
|
||||
print('Creating outbox thread ' +
|
||||
accountOutboxThreadName + '/' +
|
||||
str(self.server.outboxThreadIndex[accountOutboxThreadName]))
|
||||
self.server.outboxThread[accountOutboxThreadName][index] = \
|
||||
threadWithTrace(target=self._postToOutbox,
|
||||
args=(messageJson.copy(),
|
||||
self.server.projectVersion, None),
|
||||
daemon=True)
|
||||
print('Starting outbox thread')
|
||||
self.server.outboxThread[accountOutboxThreadName].start()
|
||||
self.server.outboxThread[accountOutboxThreadName][index].start()
|
||||
return True
|
||||
|
||||
def _updateInboxQueue(self, nickname: str, messageJson: {},
|
||||
|
|
@ -7126,19 +7151,19 @@ class PubServer(BaseHTTPRequestHandler):
|
|||
self._404()
|
||||
self.server.GETbusy = False
|
||||
return
|
||||
manualApproveFollowRequest(self.server.session,
|
||||
baseDir, httpPrefix,
|
||||
followerNickname,
|
||||
domain, port,
|
||||
followingHandle,
|
||||
self.server.federationList,
|
||||
self.server.sendThreads,
|
||||
self.server.postLog,
|
||||
self.server.cachedWebfingers,
|
||||
self.server.personCache,
|
||||
debug,
|
||||
self.server.projectVersion,
|
||||
self.server.signingPrivateKeyPem)
|
||||
manualApproveFollowRequestThread(self.server.session,
|
||||
baseDir, httpPrefix,
|
||||
followerNickname,
|
||||
domain, port,
|
||||
followingHandle,
|
||||
self.server.federationList,
|
||||
self.server.sendThreads,
|
||||
self.server.postLog,
|
||||
self.server.cachedWebfingers,
|
||||
self.server.personCache,
|
||||
debug,
|
||||
self.server.projectVersion,
|
||||
self.server.signingPrivateKeyPem)
|
||||
originPathStrAbsolute = \
|
||||
httpPrefix + '://' + domainFull + originPathStr
|
||||
if callingDomain.endswith('.onion') and onionDomain:
|
||||
|
|
@ -7284,19 +7309,19 @@ class PubServer(BaseHTTPRequestHandler):
|
|||
followingHandle = \
|
||||
handleNickname + '@' + getFullDomain(handleDomain, handlePort)
|
||||
if '@' in followingHandle:
|
||||
manualDenyFollowRequest(self.server.session,
|
||||
baseDir, httpPrefix,
|
||||
followerNickname,
|
||||
domain, port,
|
||||
followingHandle,
|
||||
self.server.federationList,
|
||||
self.server.sendThreads,
|
||||
self.server.postLog,
|
||||
self.server.cachedWebfingers,
|
||||
self.server.personCache,
|
||||
debug,
|
||||
self.server.projectVersion,
|
||||
self.server.signingPrivateKeyPem)
|
||||
manualDenyFollowRequestThread(self.server.session,
|
||||
baseDir, httpPrefix,
|
||||
followerNickname,
|
||||
domain, port,
|
||||
followingHandle,
|
||||
self.server.federationList,
|
||||
self.server.sendThreads,
|
||||
self.server.postLog,
|
||||
self.server.cachedWebfingers,
|
||||
self.server.personCache,
|
||||
debug,
|
||||
self.server.projectVersion,
|
||||
self.server.signingPrivateKeyPem)
|
||||
originPathStrAbsolute = \
|
||||
httpPrefix + '://' + domainFull + originPathStr
|
||||
if callingDomain.endswith('.onion') and onionDomain:
|
||||
|
|
@ -17197,6 +17222,7 @@ def runDaemon(listsEnabled: str,
|
|||
httpd.registration = False
|
||||
httpd.enableSharedInbox = enableSharedInbox
|
||||
httpd.outboxThread = {}
|
||||
httpd.outboxThreadIndex = {}
|
||||
httpd.newPostThread = {}
|
||||
httpd.projectVersion = projectVersion
|
||||
httpd.secureMode = secureMode
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ from utils import removeDomainPort
|
|||
from utils import getPortFromDomain
|
||||
from utils import getUserPaths
|
||||
from utils import acctDir
|
||||
from threads import threadWithTrace
|
||||
|
||||
|
||||
def manualDenyFollowRequest(session, baseDir: str,
|
||||
|
|
@ -67,6 +68,35 @@ def manualDenyFollowRequest(session, baseDir: str,
|
|||
print('Follow request from ' + denyHandle + ' was denied.')
|
||||
|
||||
|
||||
def manualDenyFollowRequestThread(session, baseDir: str,
|
||||
httpPrefix: str,
|
||||
nickname: str, domain: str, port: int,
|
||||
denyHandle: str,
|
||||
federationList: [],
|
||||
sendThreads: [], postLog: [],
|
||||
cachedWebfingers: {}, personCache: {},
|
||||
debug: bool,
|
||||
projectVersion: str,
|
||||
signingPrivateKeyPem: str) -> None:
|
||||
"""Manually deny a follow request, within a thread so that the
|
||||
user interface doesn't lag
|
||||
"""
|
||||
thr = \
|
||||
threadWithTrace(target=manualDenyFollowRequest,
|
||||
args=(session, baseDir,
|
||||
httpPrefix,
|
||||
nickname, domain, port,
|
||||
denyHandle,
|
||||
federationList,
|
||||
sendThreads, postLog,
|
||||
cachedWebfingers, personCache,
|
||||
debug,
|
||||
projectVersion,
|
||||
signingPrivateKeyPem), daemon=True)
|
||||
thr.start()
|
||||
sendThreads.append(thr)
|
||||
|
||||
|
||||
def _approveFollowerHandle(accountDir: str, approveHandle: str) -> None:
|
||||
""" Record manually approved handles so that if they unfollow and then
|
||||
re-follow later then they don't need to be manually approved again
|
||||
|
|
@ -231,3 +261,32 @@ def manualApproveFollowRequest(session, baseDir: str,
|
|||
os.remove(approveFollowsFilename + '.new')
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
|
||||
def manualApproveFollowRequestThread(session, baseDir: str,
|
||||
httpPrefix: str,
|
||||
nickname: str, domain: str, port: int,
|
||||
approveHandle: str,
|
||||
federationList: [],
|
||||
sendThreads: [], postLog: [],
|
||||
cachedWebfingers: {}, personCache: {},
|
||||
debug: bool,
|
||||
projectVersion: str,
|
||||
signingPrivateKeyPem: str) -> None:
|
||||
"""Manually approve a follow request, in a thread so as not to cause
|
||||
the UI to lag
|
||||
"""
|
||||
thr = \
|
||||
threadWithTrace(target=manualApproveFollowRequest,
|
||||
args=(session, baseDir,
|
||||
httpPrefix,
|
||||
nickname, domain, port,
|
||||
approveHandle,
|
||||
federationList,
|
||||
sendThreads, postLog,
|
||||
cachedWebfingers, personCache,
|
||||
debug,
|
||||
projectVersion,
|
||||
signingPrivateKeyPem), daemon=True)
|
||||
thr.start()
|
||||
sendThreads.append(thr)
|
||||
|
|
|
|||
Loading…
Reference in New Issue