Commit 39666ddd authored by Denis Salem's avatar Denis Salem
Browse files

WIP: Mess with parallel processing...

parent 1f438f14
......@@ -23,6 +23,8 @@ import shutil
import subprocess
import time
from multiprocessing import Process, Manager
from venc2.commands.remote import remote_copy
from venc2.datastore import DataStore
from venc2.datastore.theme import Theme
......@@ -37,7 +39,7 @@ from venc2.patterns.exceptions import MalformedPatterns
from venc2.patterns.patterns_map import PatternsMap
from venc2.patterns.processor import Processor
from venc2.patterns.processor import ProcessedString
# Initialisation of environment
start_timestamp = time.time()
datastore = None
......@@ -108,7 +110,9 @@ def init_theme(argv):
handle_malformed_patterns(e)
def setup_pattern_processor(pattern_map):
processor = Processor()
cpu_threads = len(datastore.cpu_threads_requested_entry)
processor = Processor(cpu_threads)
for pattern_name in pattern_map.non_contextual["entries"].keys():
processor.set_function(pattern_name, pattern_map.non_contextual["entries"][pattern_name])
......@@ -130,63 +134,84 @@ def setup_pattern_processor(pattern_map):
processor.blacklist.append("Escape")
return processor
def thread_process_non_contextual_entry_patterns(pattern_processor, theme, entries, cpu_thread_id):
for entry in entries:
def worker_process_non_contextual_entry_patterns(shared_data, worker_id):
for entry in shared_data["datastore"].entries[worker_id*(chunks_len):(worker_id+1)*(chunks_len)]:
shared_data["datastore"].cpu_threads_requested_entry[worker_id] = entry
print(shared_data["datastore"].cpu_threads_requested_entry)
return
#datastore, pattern_processor, theme, chunks_len
workers_count = len(datastore.cpu_threads_requested_entry)
notify("│ "+("└─ " if worker_id == workers_count - 1 else "├─ ")+messages.start_thread.format(worker_id))
for entry in datastore.entries[worker_id*(chunks_len):(worker_id+1)*(chunks_len)]:
datastore.cpu_threads_requested_entry[0] = entry
if hasattr(entry, "markup_language"):
markup_language = getattr(entry, "markup_language")
else:
markup_language = datastore.blog_configuration["markup_language"]
pattern_processor.process(entry.preview, cpu_thread_id)
pattern_processor.process(entry.preview, 0)
process_markup_language(entry.preview, markup_language)
pattern_processor.process(entry.content, cpu_thread_id)
pattern_processor.process(entry.content, 0)
process_markup_language(entry.content, markup_language, entry)
entry.html_wrapper = deepcopy(theme.entry)
pattern_processor.process(entry.html_wrapper.processed_string, cpu_thread_id)
pattern_processor.process(entry.html_wrapper.processed_string, 0)
entry.html_wrapper.processed_string.replace_needles()
entry.rss_wrapper = deepcopy(theme.rss_entry)
pattern_processor.process(entry.rss_wrapper.processed_string, cpu_thread_id)
pattern_processor.process(entry.rss_wrapper.processed_string, 0)
entry.rss_wrapper.processed_string.replace_needles()
entry.atom_wrapper = deepcopy(theme.atom_entry)
pattern_processor.process(entry.atom_wrapper.processed_string, cpu_thread_id)
pattern_processor.process(entry.atom_wrapper.processed_string, 0)
entry.atom_wrapper.processed_string.replace_needles()
def process_non_contextual_patterns(pattern_processor, theme, patterns_map):
entries = [entry for entry in datastore.get_entries()]
workers_count = 1 # len(datastore.cpu_threads_requested_entry)
try:
from threading import Thread
from multiprocessing import cpu_count
slice_len = (len(entries)//cpu_count())+1
threads = []
for i in range(0, cpu_count()):
threads.append(Thread(
target=thread_process_non_contextual_entry_patterns,
args = (
pattern_processor,
theme,
entries[i*(slice_len):(i+1)*(slice_len)],
i
chunks_len = (len(entries)//workers_count)+1
workers = []
manager = Manager()
shared_data = manager.dict({
"datastore": datastore,
"pattern_processor":pattern_processor,
"theme":theme,
"chunks_len":chunks_len,
})
if False and workers_count > 1:
for i in range(0, workers_count):
workers.append(
Process(
target=worker_process_non_contextual_entry_patterns,
args=(
shared_data,
i
)
)
))
threads[-1].start()
for thread in threads:
thread.join()
)
for worker in workers:
worker.start()
except ModuleNotFoundError:
thread_process_non_contextual_entry_patterns(
for worker in workers:
worker.join()
else:
worker_process_non_contextual_entry_patterns(
datastore,
pattern_processor,
theme,
entries,
chunks_len,
0
)
for pattern_name in patterns_map.non_contextual["entries"].keys():
pattern_processor.del_function(pattern_name)
......@@ -220,7 +245,9 @@ def export_blog(argv=list()):
notify("├─ "+messages.pre_process)
process_non_contextual_patterns(pattern_processor, theme, patterns_map)
PRINT("CUT")
exit(0)
# cleaning directory
shutil.rmtree("blog", ignore_errors=False, onerror=rm_tree_error_handler)
os.makedirs("blog")
......@@ -269,7 +296,8 @@ def export_blog(argv=list()):
def edit_and_export(argv):
global datastore
datastore = DataStore()
datastore = manager.DataStore()
if len(argv) != 1:
die(messages.missing_params.format("--edit-and-export"))
......
......@@ -23,6 +23,7 @@ import json
import os
import unidecode
from multiprocessing import cpu_count, Manager
from urllib.parse import quote as urllib_parse_quote
from venc2.datastore.configuration import get_blog_configuration
......@@ -66,13 +67,13 @@ class DataStore:
self.entries = list()
self.entries_per_archives = list()
self.entries_per_categories = list()
try:
from multiprocessing import cpu_count
self.cpu_thread_requested_entry_index = [0]*cpu_count()
except:
self.cpu_thread_requested_entry_index = [0]
try:
self.cpu_threads_requested_entry = [None]*cpu_count()
except NotImplementedError:
self.cpu_threads_requested_entry = [None]
self.max_category_weight = 1
self.categories_leaves = []
self.embed_providers = {}
......@@ -466,7 +467,7 @@ class DataStore:
self.categories_as_jsonld[path]["blogPost"].append(blog_post)
def get_chapters(self, argv):
def get_chapters(self, cpu_thread_id, argv):
key = ''.join(argv)
if not key in self.html_chapters.keys():
self.html_chapters[key] = self.build_html_chapters(argv, self.chapters_index, 0)
......@@ -532,17 +533,18 @@ class DataStore:
self.max_category_weight = value
return value
def get_generation_timestamp(self, argv):
def get_generation_timestamp(self, cpu_thread_id, argv):
if len(argv):
return datetime.datetime.strftime(self.generation_timestamp, argv[0])
else:
raise PatternMissingArguments
def get_blog_metadata(self, argv):
def get_blog_metadata(self, cpu_thread_id, argv):
# if exception is raised it will be automatically be catch by processor.
try:
return self.blog_configuration[argv[0]]
except KeyError:
raise PatternInvalidArgument(
"GetBlogMetadata",
......@@ -550,7 +552,7 @@ class DataStore:
messages.blog_has_no_metadata_like.format(argv[0])
)
def get_blog_metadata_if_exists(self, argv, ok_if_null=True):
def get_blog_metadata_if_exists(self, cpu_thread_id, argv, ok_if_null=True):
try:
value = self.blog_configuration[argv[0]]
......@@ -573,13 +575,13 @@ class DataStore:
except IndexError:
return value
def get_blog_metadata_if_not_null(self, argv):
return self.get_blog_metadata_if_exists(argv, ok_if_null=False)
def get_blog_metadata_if_not_null(self, cpu_thread_id, argv):
return self.get_blog_metadata_if_exists(cpu_thread_id, argv, ok_if_null=False)
def get_entry_metadata(self, argv):
def get_entry_metadata(self, cpu_thread_id, argv):
# if exception is raised it will be automatically be catch by processor.
try:
return str(getattr(self.entries[self.requested_entry_index], argv[0]))
return str(getattr(self.cpu_threads_requested_entry[cpu_thread_id], argv[0]))
except AttributeError:
raise PatternInvalidArgument(
......@@ -588,9 +590,9 @@ class DataStore:
messages.entry_has_no_metadata_like.format(argv[0])
)
def get_entry_metadata_if_exists(self, argv, ok_if_null=True):
def get_entry_metadata_if_exists(self, cpu_thread_id, argv, ok_if_null=True):
try:
value = str(getattr(self.entries[self.requested_entry_index], argv[0]))
value = str(getattr(self.cpu_threads_requested_entry[cpu_thread_id], argv[0]))
except AttributeError:
if len(argv) >= 3:
......@@ -630,54 +632,50 @@ class DataStore:
index += 1
for entry in (self.entries_per_archives[index].related_to[::-1] if reverse else self.entries_per_archives[index].related_to):
self.requested_entry_index = entry
yield self.entries[entry]
def get_entries(self, reverse=False):
self.requested_entry_index = 0 if not reverse else len(self.entries) - 1
for entry in (self.entries[::-1] if reverse else self.entries):
yield entry
if not reverse:
self.requested_entry_index += 1
else:
self.requested_entry_index -= 1
def get_entry_title(self, argv=list()):
title = self.entries[self.requested_entry_index].title
def get_entry_title(self, cpu_thread_id, argv=list()):
title = self.cpu_threads_requested_entry[cpu_thread_id].title
return title if title != None else str()
def get_entry_id(self, argv=list()):
return self.entries[self.requested_entry_index].id
def get_entry_year(self, argv=list()):
return self.entries[self.requested_entry_index].date.year
def get_entry_id(self, cpu_thread_id, argv=list()):
try:
return self.cpu_threads_requested_entry[cpu_thread_id].id
except AttributeError as e:
print("FAILURE IN THREAD ID", cpu_thread_id, self.cpu_threads_requested_entry, e)
raise e
def get_entry_year(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.year
def get_entry_month(self, argv=list()):
return self.entries[self.requested_entry_index].date.month
def get_entry_month(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.month
def get_entry_day(self, argv=list()):
return self.entries[self.requested_entry_index].date.day
def get_entry_day(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.day
def get_entry_hour(self, argv=list()):
return self.entries[self.requested_entry_index].date.hour
def get_entry_hour(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.hour
def get_entry_minute(self, argv=list()):
return self.entries[self.requested_entry_index].date.minute
def get_entry_minute(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.minute
def get_entry_date(self, argv=list()):
return self.entries[self.requested_entry_index].date.strftime(
def get_entry_date(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.strftime(
self.blog_configuration["date_format"] if len(argv) < 1 else argv[0]
)
def get_entry_date_url(self, argv=list()):
return self.entries[self.requested_entry_index].date.strftime(
def get_entry_date_url(self, cpu_thread_id, argv=list()):
return self.cpu_threads_requested_entry[cpu_thread_id].date.strftime(
self.blog_configuration["path"]["archives_directory_name"]
)
def get_chapter_attribute_by_index(self, argv=list()):
def get_chapter_attribute_by_index(self, cpu_thread_id, argv=list()):
if len(argv) < 2:
raise PatternMissingArguments(2, len(argv))
......@@ -702,7 +700,7 @@ class DataStore:
return self.cache_get_chapter_attribute_by_index[key]
def get_entry_attribute_by_id(self, argv=list()):
def get_entry_attribute_by_id(self, cpu_thread_id, argv=list()):
if len(argv) < 2:
raise PatternMissingArguments(2, len(argv))
......@@ -735,40 +733,40 @@ class DataStore:
return self.cache_get_entry_attribute_by_id[key]
def get_entry_url(self, argv=list()):
def get_entry_url(self, cpu_thread_id, argv=list()):
if self.blog_configuration["disable_single_entries"]:
return ''
return self.entries[self.requested_entry_index].url
return self.cpu_threads_requested_entry[cpu_thread_id].url
def get_author_name(self, argv=list()):
def get_author_name(self, cpu_thread_id, argv=list()):
return self.blog_configuration["author_name"]
def get_blog_name(self, argv=list()):
def get_blog_name(self, cpu_thread_id, argv=list()):
return self.blog_configuration["blog_name"]
def get_blog_description(self, argv=list()):
def get_blog_description(self, cpu_thread_id, argv=list()):
return self.blog_configuration["blog_description"]
def get_blog_keywords(self, argv=list()):
def get_blog_keywords(self, cpu_thread_id, argv=list()):
return self.blog_configuration["blog_keywords"]
def get_author_description(self, argv=list()):
def get_author_description(self, cpu_thread_id, argv=list()):
return self.blog_configuration["author_description"]
def get_blog_license(self, argv=list()):
def get_blog_license(self, cpu_thread_id, argv=list()):
return self.blog_configuration["license"]
def get_blog_url(self, argv=list()):
def get_blog_url(self, cpu_thread_id, argv=list()):
return self.blog_configuration["blog_url"]
def get_blog_language(self, argv=list()):
def get_blog_language(self, cpu_thread_id, argv=list()):
return self.blog_configuration["blog_language"]
def get_author_email(self, argv=list()):
def get_author_email(self, cpu_thread_id, argv=list()):
return self.blog_configuration["author_email"]
def for_blog_archives(self, argv):
def for_blog_archives(self, cpu_thread_id, argv):
key = ''.join(argv)
if not key in self.html_blog_archives.keys():
if self.blog_configuration["disable_archives"]:
......@@ -780,7 +778,7 @@ class DataStore:
return self.html_blog_archives[key]
def get_root_page(self, argv):
def get_root_page(self, cpu_thread_id, argv):
if self.root_page == None:
self.root_page = "\x1a"+self.blog_configuration["path"]["index_file_name"].format(**{"page_number":''})
......@@ -816,24 +814,26 @@ class DataStore:
return output_string + closing_node
def tree_for_entry_categories(self, argv):
def tree_for_entry_categories(self, cpu_thread_id, argv):
key = ''.join(argv)
entry = self.cpu_threads_requested_entry[cpu_thread_id]
if not key in self.entries[self.requested_entry_index].html_categories_tree.keys():
if self.blog_configuration["disable_categories"]:
self.entries[self.requested_entry_index].html_categories_tree[key] = ''
entry.html_categories_tree[key] = ''
else:
self.entries[self.requested_entry_index].html_categories_tree[key] = self.build_html_categories_tree(
entry.html_categories_tree[key] = self.build_html_categories_tree(
argv[0], #opening_node
argv[1], #opening_branch
argv[2], #closing_branch
argv[3], #closing_node
self.entries[self.requested_entry_index].categories_tree
entry.categories_tree
)
return self.entries[self.requested_entry_index].html_categories_tree[key]
return entry.html_categories_tree[key]
def tree_for_blog_categories(self, argv):
def tree_for_blog_categories(self, cpu_thread_id, argv):
key = ''.join(argv)
# compute once categories tree and deliver baked html
if not key in self.html_categories_tree.keys():
......@@ -852,7 +852,7 @@ class DataStore:
return self.html_categories_tree[key]
# TODO: NOT FINISHED YET
def for_entry_range(self, argv):
def for_entry_range(self, cpu_thread_id, argv):
if len(argv) != 3:
raise PatternMissingArguments(expected=2,got=len(argv))
......@@ -877,7 +877,7 @@ class DataStore:
if end_from <= start_from:
raise GenericMessage(messages.invalid_range.format(start_from, end_to))
entry = self.entries[self.requested_entry_index]
entry = self.cpu_threads_requested_entry[cpu_thread_id]
output = ""
#TODO: PREVENT CRASH IN CASE OF WRONG INPUTS
......@@ -886,11 +886,11 @@ class DataStore:
return output
def for_entry_metadata(self, argv):
def for_entry_metadata(self, cpu_thread_id, argv):
if len(argv) != 3:
raise PatternMissingArguments(expected=3,got=len(argv))
entry = self.entries[self.requested_entry_index]
entry = self.cpu_threads_requested_entry[cpu_thread_id]
key = ''.join(argv)
if not key in entry.html_for_metadata:
......@@ -917,16 +917,16 @@ class DataStore:
return entry.html_for_metadata[key]
def for_entry_authors(self, argv):
return self.for_entry_metadata(["authors"]+argv)
def for_entry_authors(self, cpu_thread_id, argv):
return self.for_entry_metadata(cpu_thread_id, ["authors"]+argv)
def for_entry_tags(self, argv):
return self.for_entry_metadata(["tags"]+argv)
def for_entry_tags(self, cpu_thread_id, argv):
return self.for_entry_metadata(cpu_thread_id, ["tags"]+argv)
# TODO in 2.x.x: Access {count} and {weight} from LeavesForEntrycategories by taking benefit of preprocessing.
def leaves_for_entry_categories(self, argv):
def leaves_for_entry_categories(self, cpu_thread_id, argv):
key = ''.join(argv)
entry = self.entries[self.requested_entry_index]
entry = self.cpu_threads_requested_entry[cpu_thread_id]
if not key in entry.html_categories_leaves.keys():
if self.blog_configuration["disable_categories"]:
entry.html_categories_leaves[key] = ''
......@@ -936,7 +936,7 @@ class DataStore:
return entry.html_categories_leaves[key]
def leaves_for_blog_categories(self, argv):
def leaves_for_blog_categories(self, cpu_thread_id, argv):
key = ''.join(argv)
if not key in self.html_categories_leaves.keys():
......@@ -965,7 +965,7 @@ class DataStore:
except FileNotFoundError:
return ""
def wrapper_embed_content(self, argv):
def wrapper_embed_content(self, cpu_thread_id, argv):
if len(argv) == 0:
raise PatternMissingArguments
......
......@@ -124,3 +124,4 @@ class Messages:
chapter_has_a_wrong_index = "Le format de l'index du chapitre n'est pas correct."
there_is_no_chapter_with_index = "There is no chapter with index \"{0}\"."
chapter_has_no_attribute_like = "Chapter has no attribute like \"{0}\"."
start_thread = "Start CPU thread #{0}."
......@@ -121,3 +121,4 @@ class Messages:
id_must_be_an_integer = "L'identifiant doit-être un nombre entier."
there_is_no_chapter_with_index = "Il n'y a pas de chapitre ayant l'index \"{0}\"."
chapter_has_no_attribute_like = "Le chapitre n'a pas d'attribut \"{0}\"."
start_thread = "Lancement du fil d'execution n° {0}."
......@@ -33,7 +33,7 @@ class CodeHighlight:
self._override = override
self._includes = dict()
def get_style_sheets(self, argv=list()):
def get_style_sheets(self, cpu_thread_id, argv=list()):
output = str()
for filename in self._includes.keys():
output += "<link rel=\"stylesheet\" href=\"\x1a"+filename+"\" type=\"text/css\" />\n"
......@@ -47,11 +47,11 @@ class CodeHighlight:
stream = open(os.getcwd()+"/extra/"+key,'w')
stream.write(self._includes[key])
def highlight_include(self, argv):
string = include_file([argv[2]])
return self.highlight(argv[:2]+[string], included_file=True)
def highlight_include(self, cpu_thread_id, argv):
string = include_file(cpu_thread_id, [argv[2]])
return self.highlight(cpu_thread_id, argv[:2]+[string], included_file=True)
def highlight(self, argv, included_file=False):
def highlight(self, cpu_thread_id, argv, included_file=False):
try:
name = "venc_source_"+argv[0].replace('+','Plus')
......
......@@ -25,7 +25,7 @@ from venc2.l10n import messages
arg_names=["min","max","decimal_number"]
def get_random_number(in_argv):
def get_random_number(cpu_thread_id, in_argv):
try:
mn, mx, precision = in_argv
......
......@@ -23,7 +23,7 @@ from venc2.l10n import messages
from venc2.patterns.exceptions import PatternInvalidArgument
from venc2.patterns.exceptions import PatternMissingArguments
def Latex2MathML(argv):
def Latex2MathML(cpu_thread_id, argv):
try:
tex_math_string = argv[0]
......
......@@ -32,7 +32,7 @@ from urllib.parse import urlparse
theme_includes_dependencies = []
def disable_markup(argv):
def disable_markup(cpu_thread_id, argv):
return '::'.join(argv)
def try_oembed(providers, url):
......@@ -81,17 +81,17 @@ def get_embed_content(providers, argv):
return try_oembed(providers, url)
def get_venc_version(argv):
def get_venc_version(cpu_thread_id, argv):
return venc_version
""" Need to handle missing args in case of unknown number of args """
def set_color(argv):
def set_color(cpu_thread_id, argv):
if len(argv) < 2:
raise PatternMissingArguments(expected=2, got=len(argv))
return "<span class=\"__VENC_TEXT_COLOR__\" style=\"color: "+argv[1]+";\">"+argv[0]+"</span>"
def set_style(argv):
def set_style(cpu_thread_id, argv):
ID = argv[0].strip()
CLASS = argv[1].strip()
ID = "id=\""+ID+"\"" if ID != '' else ''
......@@ -102,7 +102,7 @@ def set_style(argv):
# TODO: Must fix dirty try/except structure.
# TODO: Add explicit message about exception.
def include_file(argv, raise_error=True):