1 /*
2 * Copyright 2011 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.handler.traffic;
17
18 import static io.netty.util.internal.ObjectUtil.checkPositive;
19
20 import io.netty.buffer.ByteBuf;
21 import io.netty.buffer.ByteBufHolder;
22 import io.netty.channel.Channel;
23 import io.netty.channel.ChannelDuplexHandler;
24 import io.netty.channel.ChannelConfig;
25 import io.netty.channel.ChannelHandlerContext;
26 import io.netty.channel.ChannelOutboundBuffer;
27 import io.netty.channel.ChannelPromise;
28 import io.netty.channel.FileRegion;
29 import io.netty.util.Attribute;
30 import io.netty.util.AttributeKey;
31 import io.netty.util.ReferenceCountUtil;
32 import io.netty.util.internal.logging.InternalLogger;
33 import io.netty.util.internal.logging.InternalLoggerFactory;
34
35 import java.util.concurrent.TimeUnit;
36
37 /**
38 * <p>AbstractTrafficShapingHandler allows to limit the global bandwidth
39 * (see {@link GlobalTrafficShapingHandler}) or per session
40 * bandwidth (see {@link ChannelTrafficShapingHandler}), as traffic shaping.
41 * It allows you to implement an almost real time monitoring of the bandwidth using
42 * the monitors from {@link TrafficCounter} that will call back every checkInterval
43 * the method doAccounting of this handler.</p>
44 *
45 * <p>If you want for any particular reasons to stop the monitoring (accounting) or to change
46 * the read/write limit or the check interval, several methods allow that for you:</p>
47 * <ul>
48 * <li><tt>configure</tt> allows you to change read or write limits, or the checkInterval</li>
49 * <li><tt>getTrafficCounter</tt> allows you to have access to the TrafficCounter and so to stop
50 * or start the monitoring, to change the checkInterval directly, or to have access to its values.</li>
51 * </ul>
52 */
53 public abstract class AbstractTrafficShapingHandler extends ChannelDuplexHandler {
54 private static final InternalLogger logger =
55 InternalLoggerFactory.getInstance(AbstractTrafficShapingHandler.class);
56 /**
57 * Default delay between two checks: 1s
58 */
59 public static final long DEFAULT_CHECK_INTERVAL = 1000;
60
61 /**
62 * Default max delay in case of traffic shaping
63 * (during which no communication will occur).
64 * Shall be less than TIMEOUT. Here half of "standard" 30s
65 */
66 public static final long DEFAULT_MAX_TIME = 15000;
67
68 /**
69 * Default max size to not exceed in buffer (write only).
70 */
71 static final long DEFAULT_MAX_SIZE = 4 * 1024 * 1024L;
72
73 /**
74 * Default minimal time to wait: 10ms
75 */
76 static final long MINIMAL_WAIT = 10;
77
78 /**
79 * Traffic Counter
80 */
81 protected TrafficCounter trafficCounter;
82
83 /**
84 * Limit in B/s to apply to write
85 */
86 private volatile long writeLimit;
87
88 /**
89 * Limit in B/s to apply to read
90 */
91 private volatile long readLimit;
92
93 /**
94 * Max delay in wait
95 */
96 protected volatile long maxTime = DEFAULT_MAX_TIME; // default 15 s
97
98 /**
99 * Delay between two performance snapshots
100 */
101 protected volatile long checkInterval = DEFAULT_CHECK_INTERVAL; // default 1 s
102
103 static final AttributeKey<Boolean> READ_SUSPENDED = AttributeKey
104 .valueOf(AbstractTrafficShapingHandler.class.getName() + ".READ_SUSPENDED");
105 static final AttributeKey<Runnable> REOPEN_TASK = AttributeKey.valueOf(AbstractTrafficShapingHandler.class
106 .getName() + ".REOPEN_TASK");
107
108 /**
109 * Max time to delay before proposing to stop writing new objects from next handlers
110 */
111 volatile long maxWriteDelay = 4 * DEFAULT_CHECK_INTERVAL; // default 4 s
112 /**
113 * Max size in the list before proposing to stop writing new objects from next handlers
114 */
115 volatile long maxWriteSize = DEFAULT_MAX_SIZE; // default 4MB
116
117 /**
118 * Rank in UserDefinedWritability (1 for Channel, 2 for Global TrafficShapingHandler).
119 * Set in final constructor. Must be between 1 and 31
120 */
121 final int userDefinedWritabilityIndex;
122
123 /**
124 * Default value for Channel UserDefinedWritability index
125 */
126 static final int CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 1;
127
128 /**
129 * Default value for Global UserDefinedWritability index
130 */
131 static final int GLOBAL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 2;
132
133 /**
134 * Default value for GlobalChannel UserDefinedWritability index
135 */
136 static final int GLOBALCHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX = 3;
137
138 /**
139 * @param newTrafficCounter
140 * the TrafficCounter to set
141 */
142 void setTrafficCounter(TrafficCounter newTrafficCounter) {
143 trafficCounter = newTrafficCounter;
144 }
145
146 /**
147 * @return the index to be used by the TrafficShapingHandler to manage the user defined writability.
148 * For Channel TSH it is defined as {@value #CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX},
149 * for Global TSH it is defined as {@value #GLOBAL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX},
150 * for GlobalChannel TSH it is defined as
151 * {@value #GLOBALCHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX}.
152 */
153 protected int userDefinedWritabilityIndex() {
154 return CHANNEL_DEFAULT_USER_DEFINED_WRITABILITY_INDEX;
155 }
156
157 /**
158 * @param writeLimit
159 * 0 or a limit in bytes/s
160 * @param readLimit
161 * 0 or a limit in bytes/s
162 * @param checkInterval
163 * The delay between two computations of performances for
164 * channels or 0 if no stats are to be computed.
165 * @param maxTime
166 * The maximum delay to wait in case of traffic excess.
167 * Must be positive.
168 */
169 protected AbstractTrafficShapingHandler(long writeLimit, long readLimit, long checkInterval, long maxTime) {
170 this.maxTime = checkPositive(maxTime, "maxTime");
171
172 userDefinedWritabilityIndex = userDefinedWritabilityIndex();
173 this.writeLimit = writeLimit;
174 this.readLimit = readLimit;
175 this.checkInterval = checkInterval;
176 }
177
178 /**
179 * Constructor using default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
180 * @param writeLimit
181 * 0 or a limit in bytes/s
182 * @param readLimit
183 * 0 or a limit in bytes/s
184 * @param checkInterval
185 * The delay between two computations of performances for
186 * channels or 0 if no stats are to be computed.
187 */
188 protected AbstractTrafficShapingHandler(long writeLimit, long readLimit, long checkInterval) {
189 this(writeLimit, readLimit, checkInterval, DEFAULT_MAX_TIME);
190 }
191
192 /**
193 * Constructor using default Check Interval value of {@value #DEFAULT_CHECK_INTERVAL} ms and
194 * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
195 *
196 * @param writeLimit
197 * 0 or a limit in bytes/s
198 * @param readLimit
199 * 0 or a limit in bytes/s
200 */
201 protected AbstractTrafficShapingHandler(long writeLimit, long readLimit) {
202 this(writeLimit, readLimit, DEFAULT_CHECK_INTERVAL, DEFAULT_MAX_TIME);
203 }
204
205 /**
206 * Constructor using NO LIMIT, default Check Interval value of {@value #DEFAULT_CHECK_INTERVAL} ms and
207 * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
208 */
209 protected AbstractTrafficShapingHandler() {
210 this(0, 0, DEFAULT_CHECK_INTERVAL, DEFAULT_MAX_TIME);
211 }
212
213 /**
214 * Constructor using NO LIMIT and
215 * default max time as delay allowed value of {@value #DEFAULT_MAX_TIME} ms.
216 *
217 * @param checkInterval
218 * The delay between two computations of performances for
219 * channels or 0 if no stats are to be computed.
220 */
221 protected AbstractTrafficShapingHandler(long checkInterval) {
222 this(0, 0, checkInterval, DEFAULT_MAX_TIME);
223 }
224
225 /**
226 * Change the underlying limitations and check interval.
227 * <p>Note the change will be taken as best effort, meaning
228 * that all already scheduled traffics will not be
229 * changed, but only applied to new traffics.</p>
230 * <p>So the expected usage of this method is to be used not too often,
231 * accordingly to the traffic shaping configuration.</p>
232 *
233 * @param newWriteLimit The new write limit (in bytes)
234 * @param newReadLimit The new read limit (in bytes)
235 * @param newCheckInterval The new check interval (in milliseconds)
236 */
237 public void configure(long newWriteLimit, long newReadLimit,
238 long newCheckInterval) {
239 configure(newWriteLimit, newReadLimit);
240 configure(newCheckInterval);
241 }
242
243 /**
244 * Change the underlying limitations.
245 * <p>Note the change will be taken as best effort, meaning
246 * that all already scheduled traffics will not be
247 * changed, but only applied to new traffics.</p>
248 * <p>So the expected usage of this method is to be used not too often,
249 * accordingly to the traffic shaping configuration.</p>
250 *
251 * @param newWriteLimit The new write limit (in bytes)
252 * @param newReadLimit The new read limit (in bytes)
253 */
254 public void configure(long newWriteLimit, long newReadLimit) {
255 writeLimit = newWriteLimit;
256 readLimit = newReadLimit;
257 if (trafficCounter != null) {
258 trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
259 }
260 }
261
262 /**
263 * Change the check interval.
264 *
265 * @param newCheckInterval The new check interval (in milliseconds)
266 */
267 public void configure(long newCheckInterval) {
268 checkInterval = newCheckInterval;
269 if (trafficCounter != null) {
270 trafficCounter.configure(checkInterval);
271 }
272 }
273
274 /**
275 * @return the writeLimit
276 */
277 public long getWriteLimit() {
278 return writeLimit;
279 }
280
281 /**
282 * <p>Note the change will be taken as best effort, meaning
283 * that all already scheduled traffics will not be
284 * changed, but only applied to new traffics.</p>
285 * <p>So the expected usage of this method is to be used not too often,
286 * accordingly to the traffic shaping configuration.</p>
287 *
288 * @param writeLimit the writeLimit to set
289 */
290 public void setWriteLimit(long writeLimit) {
291 this.writeLimit = writeLimit;
292 if (trafficCounter != null) {
293 trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
294 }
295 }
296
297 /**
298 * @return the readLimit
299 */
300 public long getReadLimit() {
301 return readLimit;
302 }
303
304 /**
305 * <p>Note the change will be taken as best effort, meaning
306 * that all already scheduled traffics will not be
307 * changed, but only applied to new traffics.</p>
308 * <p>So the expected usage of this method is to be used not too often,
309 * accordingly to the traffic shaping configuration.</p>
310 *
311 * @param readLimit the readLimit to set
312 */
313 public void setReadLimit(long readLimit) {
314 this.readLimit = readLimit;
315 if (trafficCounter != null) {
316 trafficCounter.resetAccounting(TrafficCounter.milliSecondFromNano());
317 }
318 }
319
320 /**
321 * @return the checkInterval
322 */
323 public long getCheckInterval() {
324 return checkInterval;
325 }
326
327 /**
328 * @param checkInterval the interval in ms between each step check to set, default value being 1000 ms.
329 */
330 public void setCheckInterval(long checkInterval) {
331 this.checkInterval = checkInterval;
332 if (trafficCounter != null) {
333 trafficCounter.configure(checkInterval);
334 }
335 }
336
337 /**
338 * <p>Note the change will be taken as best effort, meaning
339 * that all already scheduled traffics will not be
340 * changed, but only applied to new traffics.</p>
341 * <p>So the expected usage of this method is to be used not too often,
342 * accordingly to the traffic shaping configuration.</p>
343 *
344 * @param maxTime
345 * Max delay in wait, shall be less than TIME OUT in related protocol.
346 * Must be positive.
347 */
348 public void setMaxTimeWait(long maxTime) {
349 this.maxTime = checkPositive(maxTime, "maxTime");
350 }
351
352 /**
353 * @return the max delay in wait to prevent TIME OUT
354 */
355 public long getMaxTimeWait() {
356 return maxTime;
357 }
358
359 /**
360 * @return the maxWriteDelay
361 */
362 public long getMaxWriteDelay() {
363 return maxWriteDelay;
364 }
365
366 /**
367 * <p>Note the change will be taken as best effort, meaning
368 * that all already scheduled traffics will not be
369 * changed, but only applied to new traffics.</p>
370 * <p>So the expected usage of this method is to be used not too often,
371 * accordingly to the traffic shaping configuration.</p>
372 *
373 * @param maxWriteDelay the maximum Write Delay in ms in the buffer allowed before write suspension is set.
374 * Must be positive.
375 */
376 public void setMaxWriteDelay(long maxWriteDelay) {
377 this.maxWriteDelay = checkPositive(maxWriteDelay, "maxWriteDelay");
378 }
379
380 /**
381 * @return the maxWriteSize default being {@value #DEFAULT_MAX_SIZE} bytes.
382 */
383 public long getMaxWriteSize() {
384 return maxWriteSize;
385 }
386
387 /**
388 * <p>Note that this limit is a best effort on memory limitation to prevent Out Of
389 * Memory Exception. To ensure it works, the handler generating the write should
390 * use one of the way provided by Netty to handle the capacity:</p>
391 * <p>- the {@code Channel.isWritable()} property and the corresponding
392 * {@code channelWritabilityChanged()}</p>
393 * <p>- the {@code ChannelFuture.addListener(new GenericFutureListener())}</p>
394 *
395 * @param maxWriteSize the maximum Write Size allowed in the buffer
396 * per channel before write suspended is set,
397 * default being {@value #DEFAULT_MAX_SIZE} bytes.
398 */
399 public void setMaxWriteSize(long maxWriteSize) {
400 this.maxWriteSize = maxWriteSize;
401 }
402
403 /**
404 * Called each time the accounting is computed from the TrafficCounters.
405 * This method could be used for instance to implement almost real time accounting.
406 *
407 * @param counter
408 * the TrafficCounter that computes its performance
409 */
410 protected void doAccounting(TrafficCounter counter) {
411 // NOOP by default
412 }
413
414 /**
415 * Class to implement setReadable at fix time
416 */
417 static final class ReopenReadTimerTask implements Runnable {
418 final ChannelHandlerContext ctx;
419 ReopenReadTimerTask(ChannelHandlerContext ctx) {
420 this.ctx = ctx;
421 }
422
423 @Override
424 public void run() {
425 Channel channel = ctx.channel();
426 ChannelConfig config = channel.config();
427 if (!config.isAutoRead() && isHandlerActive(ctx)) {
428 // If AutoRead is False and Active is True, user make a direct setAutoRead(false)
429 // Then Just reset the status
430 if (logger.isDebugEnabled()) {
431 logger.debug("Not unsuspend: " + config.isAutoRead() + ':' +
432 isHandlerActive(ctx));
433 }
434 channel.attr(READ_SUSPENDED).set(false);
435 } else {
436 // Anything else allows the handler to reset the AutoRead
437 if (logger.isDebugEnabled()) {
438 if (config.isAutoRead() && !isHandlerActive(ctx)) {
439 if (logger.isDebugEnabled()) {
440 logger.debug("Unsuspend: " + config.isAutoRead() + ':' +
441 isHandlerActive(ctx));
442 }
443 } else {
444 if (logger.isDebugEnabled()) {
445 logger.debug("Normal unsuspend: " + config.isAutoRead() + ':'
446 + isHandlerActive(ctx));
447 }
448 }
449 }
450 channel.attr(READ_SUSPENDED).set(false);
451 config.setAutoRead(true);
452 channel.read();
453 }
454 if (logger.isDebugEnabled()) {
455 logger.debug("Unsuspend final status => " + config.isAutoRead() + ':'
456 + isHandlerActive(ctx));
457 }
458 }
459 }
460
461 /**
462 * Release the Read suspension
463 */
464 void releaseReadSuspended(ChannelHandlerContext ctx) {
465 Channel channel = ctx.channel();
466 channel.attr(READ_SUSPENDED).set(false);
467 channel.config().setAutoRead(true);
468 }
469
470 @Override
471 public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
472 long size = calculateSize(msg);
473 long now = TrafficCounter.milliSecondFromNano();
474 if (size > 0) {
475 // compute the number of ms to wait before reopening the channel
476 long wait = trafficCounter.readTimeToWait(size, readLimit, maxTime, now);
477 wait = checkWaitReadTime(ctx, wait, now);
478 if (wait >= MINIMAL_WAIT) { // At least 10ms seems a minimal
479 // time in order to try to limit the traffic
480 // Only AutoRead AND HandlerActive True means Context Active
481 Channel channel = ctx.channel();
482 ChannelConfig config = channel.config();
483 if (logger.isDebugEnabled()) {
484 logger.debug("Read suspend: " + wait + ':' + config.isAutoRead() + ':'
485 + isHandlerActive(ctx));
486 }
487 if (config.isAutoRead() && isHandlerActive(ctx)) {
488 config.setAutoRead(false);
489 channel.attr(READ_SUSPENDED).set(true);
490 // Create a Runnable to reactive the read if needed. If one was create before it will just be
491 // reused to limit object creation
492 Attribute<Runnable> attr = channel.attr(REOPEN_TASK);
493 Runnable reopenTask = attr.get();
494 if (reopenTask == null) {
495 reopenTask = new ReopenReadTimerTask(ctx);
496 attr.set(reopenTask);
497 }
498 ctx.executor().schedule(reopenTask, wait, TimeUnit.MILLISECONDS);
499 if (logger.isDebugEnabled()) {
500 logger.debug("Suspend final status => " + config.isAutoRead() + ':'
501 + isHandlerActive(ctx) + " will reopened at: " + wait);
502 }
503 }
504 }
505 }
506 informReadOperation(ctx, now);
507 ctx.fireChannelRead(msg);
508 }
509
510 @Override
511 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
512 Channel channel = ctx.channel();
513 if (channel.hasAttr(REOPEN_TASK)) {
514 //release the reopen task
515 channel.attr(REOPEN_TASK).set(null);
516 }
517 super.handlerRemoved(ctx);
518 }
519
520 /**
521 * Method overridden in GTSH to take into account specific timer for the channel.
522 * @param wait the wait delay computed in ms
523 * @param now the relative now time in ms
524 * @return the wait to use according to the context
525 */
526 long checkWaitReadTime(final ChannelHandlerContext ctx, long wait, final long now) {
527 // no change by default
528 return wait;
529 }
530
531 /**
532 * Method overridden in GTSH to take into account specific timer for the channel.
533 * @param now the relative now time in ms
534 */
535 void informReadOperation(final ChannelHandlerContext ctx, final long now) {
536 // default noop
537 }
538
539 protected static boolean isHandlerActive(ChannelHandlerContext ctx) {
540 Boolean suspended = ctx.channel().attr(READ_SUSPENDED).get();
541 return suspended == null || Boolean.FALSE.equals(suspended);
542 }
543
544 @Override
545 public void read(ChannelHandlerContext ctx) {
546 if (isHandlerActive(ctx)) {
547 // For Global Traffic (and Read when using EventLoop in pipeline) : check if READ_SUSPENDED is False
548 ctx.read();
549 }
550 }
551
552 @Override
553 public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise)
554 throws Exception {
555 long size = calculateSize(msg);
556 long now = TrafficCounter.milliSecondFromNano();
557 if (size > 0) {
558 // compute the number of ms to wait before continue with the channel
559 long wait = trafficCounter.writeTimeToWait(size, writeLimit, maxTime, now);
560 if (wait >= MINIMAL_WAIT) {
561 if (logger.isDebugEnabled()) {
562 logger.debug("Write suspend: " + wait + ':' + ctx.channel().config().isAutoRead() + ':'
563 + isHandlerActive(ctx));
564 }
565 submitWrite(ctx, msg, size, wait, now, promise);
566 return;
567 }
568 }
569 // to maintain order of write
570 submitWrite(ctx, msg, size, 0, now, promise);
571 }
572
573 @Deprecated
574 protected void submitWrite(final ChannelHandlerContext ctx, final Object msg,
575 final long delay, final ChannelPromise promise) {
576 submitWrite(ctx, msg, calculateSize(msg),
577 delay, TrafficCounter.milliSecondFromNano(), promise);
578 }
579
580 abstract void submitWrite(
581 ChannelHandlerContext ctx, Object msg, long size, long delay, long now, ChannelPromise promise);
582
583 /**
584 * Releases the given {@code msg} and fails the given {@code promise} with the supplied {@code cause}.
585 */
586 static void releaseAndFailQueuedWrite(Object msg, ChannelPromise promise, Throwable cause) {
587 ReferenceCountUtil.safeRelease(msg);
588 promise.tryFailure(cause);
589 }
590
591 @Override
592 public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
593 setUserDefinedWritability(ctx, true);
594 super.channelRegistered(ctx);
595 }
596
597 void setUserDefinedWritability(ChannelHandlerContext ctx, boolean writable) {
598 ChannelOutboundBuffer cob = ctx.channel().unsafe().outboundBuffer();
599 if (cob != null) {
600 cob.setUserDefinedWritability(userDefinedWritabilityIndex, writable);
601 }
602 }
603
604 /**
605 * Check the writability according to delay and size for the channel.
606 * Set if necessary setUserDefinedWritability status.
607 * @param delay the computed delay
608 * @param queueSize the current queueSize
609 */
610 void checkWriteSuspend(ChannelHandlerContext ctx, long delay, long queueSize) {
611 if (queueSize > maxWriteSize || delay > maxWriteDelay) {
612 setUserDefinedWritability(ctx, false);
613 }
614 }
615 /**
616 * Explicitly release the Write suspended status.
617 */
618 void releaseWriteSuspended(ChannelHandlerContext ctx) {
619 setUserDefinedWritability(ctx, true);
620 }
621
622 /**
623 * @return the current TrafficCounter (if
624 * channel is still connected)
625 */
626 public TrafficCounter trafficCounter() {
627 return trafficCounter;
628 }
629
630 @Override
631 public String toString() {
632 StringBuilder builder = new StringBuilder(290)
633 .append("TrafficShaping with Write Limit: ").append(writeLimit)
634 .append(" Read Limit: ").append(readLimit)
635 .append(" CheckInterval: ").append(checkInterval)
636 .append(" maxDelay: ").append(maxWriteDelay)
637 .append(" maxSize: ").append(maxWriteSize)
638 .append(" and Counter: ");
639 if (trafficCounter != null) {
640 builder.append(trafficCounter);
641 } else {
642 builder.append("none");
643 }
644 return builder.toString();
645 }
646
647 /**
648 * Calculate the size of the given {@link Object}.
649 *
650 * This implementation supports {@link ByteBuf}, {@link ByteBufHolder} and {@link FileRegion}.
651 * Sub-classes may override this.
652 * @param msg the msg for which the size should be calculated.
653 * @return size the size of the msg or {@code -1} if unknown.
654 */
655 protected long calculateSize(Object msg) {
656 if (msg instanceof ByteBuf) {
657 return ((ByteBuf) msg).readableBytes();
658 }
659 if (msg instanceof ByteBufHolder) {
660 return ((ByteBufHolder) msg).content().readableBytes();
661 }
662 if (msg instanceof FileRegion) {
663 return ((FileRegion) msg).count();
664 }
665 return -1;
666 }
667 }