CollapsedForwarding.cc
Go to the documentation of this file.
1 /*
2  * Copyright (C) 1996-2017 The Squid Software Foundation and contributors
3  *
4  * Squid software is distributed under GPLv2+ license and includes
5  * contributions from numerous individuals and organizations.
6  * Please see the COPYING and CONTRIBUTORS files for details.
7  */
8 
9 /* DEBUG: section 17 Request Forwarding */
10 
11 #include "squid.h"
12 #include "CollapsedForwarding.h"
13 #include "globals.h"
14 #include "ipc/mem/Segment.h"
15 #include "ipc/Messages.h"
16 #include "ipc/Port.h"
17 #include "ipc/TypedMsgHdr.h"
18 #include "MemObject.h"
19 #include "SquidConfig.h"
20 #include "Store.h"
21 #include "store_key_md5.h"
22 #include "tools.h"
23 
25 static const char *const ShmLabel = "cf";
27 // TODO: make configurable or compute from squid.conf settings if possible
28 static const int QueueCapacity = 1024;
29 
30 std::unique_ptr<CollapsedForwarding::Queue> CollapsedForwarding::queue;
31 
34 {
35 public:
37 
38 public:
39  int sender;
40 
43 };
44 
45 // CollapsedForwarding
46 
47 void
49 {
50  Must(!queue.get());
51  if (UsingSmp() && IamWorkerProcess())
52  queue.reset(new Queue(ShmLabel, KidIdentifier));
53 }
54 
55 void
57 {
58  if (!queue.get())
59  return;
60 
61  if (!e.mem_obj || e.mem_obj->xitTable.index < 0 ||
63  debugs(17, 7, "nobody reads " << e);
64  return;
65  }
66 
68  msg.sender = KidIdentifier;
69  msg.xitIndex = e.mem_obj->xitTable.index;
70 
71  debugs(17, 5, e << " to " << Config.workers << "-1 workers");
72 
73  // TODO: send only to workers who are waiting for data
74  for (int workerId = 1; workerId <= Config.workers; ++workerId) {
75  try {
76  if (workerId != KidIdentifier && queue->push(workerId, msg))
77  Notify(workerId);
78  } catch (const Queue::Full &) {
79  debugs(17, DBG_IMPORTANT, "ERROR: Collapsed forwarding " <<
80  "queue overflow for kid" << workerId <<
81  " at " << queue->outSize(workerId) << " items");
82  // TODO: grow queue size
83  }
84  }
85 }
86 
87 void
88 CollapsedForwarding::Notify(const int workerId)
89 {
90  // TODO: Count and report the total number of notifications, pops, pushes.
91  debugs(17, 7, "to kid" << workerId);
92  Ipc::TypedMsgHdr msg;
94  msg.putInt(KidIdentifier);
96  Ipc::SendMessage(addr, msg);
97 }
98 
99 void
100 CollapsedForwarding::HandleNewData(const char *const when)
101 {
102  debugs(17, 4, "popping all " << when);
104  int workerId;
105  int poppedCount = 0;
106  while (queue->pop(workerId, msg)) {
107  debugs(17, 3, "message from kid" << workerId);
108  if (workerId != msg.sender) {
109  debugs(17, DBG_IMPORTANT, "mismatching kid IDs: " << workerId <<
110  " != " << msg.sender);
111  }
112 
113  debugs(17, 7, "handling entry " << msg.xitIndex << " in transients_map");
115  debugs(17, 7, "handled entry " << msg.xitIndex << " in transients_map");
116 
117  // XXX: stop and schedule an async call to continue
118  ++poppedCount;
119  assert(poppedCount < SQUID_MAXFD);
120  }
121 }
122 
123 void
125 {
126  const int from = msg.getInt();
127  debugs(17, 7, "from " << from);
128  assert(queue.get());
129  queue->clearReaderSignal(from);
130  HandleNewData("after notification");
131 }
132 
135 {
136 public:
137  /* RegisteredRunner API */
139  virtual ~CollapsedForwardingRr();
140 
141 protected:
142  virtual void create();
143  virtual void open();
144 
145 private:
147 };
148 
150 
152 {
153  Must(!owner);
155  sizeof(CollapsedForwardingMsg),
156  QueueCapacity);
157 }
158 
160 {
162 }
163 
165 {
166  delete owner;
167 }
168 
#define assert(EX)
Definition: assert.h:17
static const char *const ShmLabel
shared memory segment path to use for CollapsedForwarding queue
initializes shared queue used by CollapsedForwarding
void putInt(int n)
store an integer
Definition: TypedMsgHdr.cc:109
static void Notify(const int workerId)
kick worker with empty IPC queue
Controller & Root()
safely access controller singleton
Definition: Controller.cc:619
XitTable xitTable
current [shared] memory caching state for the entry
Definition: MemObject.h:138
bool IamWorkerProcess()
whether the current process handles HTTP transactions and such
Definition: stub_tools.cc:49
static void Broadcast(const StoreEntry &e)
notify other workers about changes in entry state (e.g., new data)
int32_t index
entry position inside the in-transit table
Definition: MemObject.h:135
static void HandleNotification(const Ipc::TypedMsgHdr &msg)
handle queue push notifications from worker or disker
int transientReaders(const StoreEntry &) const
number of the transient entry readers some time ago
Definition: Controller.cc:445
static const int QueueCapacity
a single worker-to-worker queue capacity
#define debugs(SECTION, LEVEL, CONTENT)
Definition: Debug.h:123
virtual void create()
called when the runner should create a new memory segment
#define DBG_IMPORTANT
Definition: Debug.h:45
static String MakeAddr(const char *proccessLabel, int id)
calculates IPC message address for strand id of processLabel type
Definition: Port.cc:51
static void HandleNewData(const char *const when)
handle new data messages in IPC queue
int sender
kid ID of sending process
void * addr
Definition: membanger.c:46
Ipc::MultiQueue::Owner * owner
sfileno xitIndex
transients index, so that workers can find [private] entries to sync
signed_int32_t sfileno
Definition: forward.h:22
int getInt() const
load an integer
Definition: TypedMsgHdr.cc:101
MemObject * mem_obj
Definition: Store.h:162
void SendMessage(const String &toAddress, const TypedMsgHdr &message)
Definition: UdsOp.cc:188
bool UsingSmp()
Whether there should be more than one worker process running.
Definition: tools.cc:658
#define Must(cond)
Definition: TextException.h:89
const char strandAddrLabel[]
strand's listening address unique label
Definition: Port.cc:22
static Owner * Init(const String &id, const int processCount, const int processIdOffset, const unsigned int maxItemSize, const int capacity)
Definition: Queue.cc:342
RunnerRegistrationEntry(CollapsedForwardingRr)
int KidIdentifier
struct msghdr with a known type, fixed-size I/O and control buffers
Definition: TypedMsgHdr.h:31
static void Init()
open shared memory segment
static std::unique_ptr< Queue > queue
IPC queue.
void setType(int aType)
sets message type; use MessageType enum
Definition: TypedMsgHdr.cc:90
class SquidConfig Config
Definition: SquidConfig.cc:12
#define NULL
Definition: types.h:166
void syncCollapsed(const sfileno)
Update local intransit entry after changes made by appending worker.
Definition: Controller.cc:529

 

Introduction

Documentation

Support

Miscellaneous

Web Site Translations

Mirrors