1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
#pragma once
#include "search/processor_factory.hpp"
#include "search/result.hpp"
#include "search/search_params.hpp"
#include "search/suggest.hpp"
#include "indexer/categories_holder.hpp"
#include "geometry/rect2d.hpp"
#include "coding/reader.hpp"
#include "base/macros.hpp"
#include "base/mutex.hpp"
#include "base/thread.hpp"
#include "std/atomic.hpp"
#include "std/condition_variable.hpp"
#include "std/function.hpp"
#include "std/mutex.hpp"
#include "std/queue.hpp"
#include "std/string.hpp"
#include "std/unique_ptr.hpp"
#include "std/vector.hpp"
#include "std/weak_ptr.hpp"
class Index;
namespace storage
{
class CountryInfoGetter;
}
namespace search
{
class EngineData;
class Processor;
// This class is used as a reference to a search processor in the
// SearchEngine's queue. It's only possible to cancel a search
// request via this reference.
//
// NOTE: this class is thread-safe.
class ProcessorHandle
{
public:
ProcessorHandle();
// Cancels processor this handle points to.
void Cancel();
private:
friend class Engine;
// Attaches the handle to a |processor|. If there was or will be a
// cancel signal, this signal will be propagated to |processor|.
// This method is called only once, when search engine starts
// the processor this handle corresponds to.
void Attach(Processor & processor);
// Detaches handle from a processor. This method is called only
// once, when search engine completes processing of the query
// that this handle corresponds to.
void Detach();
Processor * m_processor;
bool m_cancelled;
mutex m_mu;
DISALLOW_COPY_AND_MOVE(ProcessorHandle);
};
// This class is a wrapper around thread which processes search
// queries one by one.
//
// NOTE: this class is thread safe.
class Engine
{
public:
struct Params
{
Params();
Params(string const & locale, size_t numThreads);
string m_locale;
// This field controls number of threads SearchEngine will create
// to process queries. Use this field wisely as large values may
// negatively affect performance due to false sharing.
size_t m_numThreads;
};
// Doesn't take ownership of index and categories.
Engine(Index & index, CategoriesHolder const & categories,
storage::CountryInfoGetter const & infoGetter, unique_ptr<ProcessorFactory> factory,
Params const & params);
~Engine();
// Posts search request to the queue and returns its handle.
weak_ptr<ProcessorHandle> Search(SearchParams const & params, m2::RectD const & viewport);
// Posts request to support old format to the queue.
void SetSupportOldFormat(bool support);
// Sets default locale on all query processors.
void SetLocale(string const & locale);
// Posts request to clear caches to the queue.
void ClearCaches();
private:
struct Message
{
using TFn = function<void(Processor & processor)>;
enum Type
{
TYPE_TASK,
TYPE_BROADCAST
};
Message(Type type, TFn && fn) : m_type(type), m_fn(move(fn)) {}
void operator()(Processor & processor) { m_fn(processor); }
Type m_type;
TFn m_fn;
};
// alignas() is used here to prevent false-sharing between different
// threads.
struct alignas(64 /* the most common cache-line size */) Context
{
// This field *CAN* be accessed by other threads, so |m_mu| must
// be taken before access this queue. Messages are ordered here
// by a timestamp and all timestamps are less than timestamps in
// the global |m_messages| queue.
queue<Message> m_messages;
// This field is thread-specific and *CAN NOT* be accessed by
// other threads.
unique_ptr<Processor> m_processor;
};
// *ALL* following methods are executed on the m_threads threads.
// This method executes tasks from a common pool (|tasks|) in a FIFO
// manner. |broadcast| contains per-thread tasks, but nevertheless
// all necessary synchronization primitives must be used to access
// |tasks| and |broadcast|.
void MainLoop(Context & context);
template <typename... TArgs>
void PostMessage(TArgs &&... args);
void DoSearch(SearchParams const & params, m2::RectD const & viewport,
shared_ptr<ProcessorHandle> handle, Processor & processor);
vector<Suggest> m_suggests;
bool m_shutdown;
mutex m_mu;
condition_variable m_cv;
queue<Message> m_messages;
vector<Context> m_contexts;
vector<threads::SimpleThread> m_threads;
};
} // namespace search
|
